[Improve] Harden the Discord gateway delivery and leadership paths (#410)

* Harden the Discord gateway delivery and leadership paths

Follow-ups from the PR #268 reliability review:

- surface the dead-letter stream: quarantine depth is tracked in gateway
  status and shown as a failing diagnostic row in Discord settings, so
  undeliverable events are visible instead of accumulating silently
- bound the durable streams: the inbound stream and dead-letter stream
  get approximate MAXLEN caps (DISCORD_GATEWAY_INBOUND_MAX_ENTRIES,
  DISCORD_GATEWAY_DEAD_LETTER_MAX_ENTRIES), and capacity pressure is
  reported in status before shedding starts
- tolerate one transient Redis error on lease renewal: the 30s lease
  leaves room for two 10s renew intervals, so only a definitive
  ownership loss or a second consecutive failure tears down the Gateway
  connection (renewDetailed distinguishes lost from error)
- restart the delivery backoff after a long healthy run instead of
  ratcheting toward the lifetime maximum forever
- report supervisor death: the embedded supervisor takes an onFatal
  hook, wired to Sentry in the BullMQ host, and stamps the shared
  status so a silently-stopped Discord ingestion pages and shows in
  diagnostics

The lease-renewal tolerance is covered indirectly (renewDetailed
contract + simple streak logic); there is no service-level harness to
drive the renew timer directly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Address gateway review findings: attempts-hash leak and erasable pressure warning

- pruneOrphanedAttempts deletes retry counters whose stream entries were
  shed by the MAXLEN cap (acknowledge/quarantine clean up their own);
  runs on the status refresh timer so the attempts hash stays bounded
  with the stream during a long outage
- the capacity pressure warning moves from lastError — which every
  successful delivery clears — to a dedicated capacityWarning status
  field whose only writer is the refresh timer, and Discord settings
  shows it as an Event backlog diagnostic row

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Snapshot attempt counters before reading the oldest stream entry

pruneOrphanedAttempts read the stream first: an event enqueued and
attempted between XRANGE (empty) and HKEYS put a live counter in the
deletion set, resetting its retry accounting past the configured
maximum. Snapshotting the tracked ids first means a counter recorded
after the snapshot can never be pruned; ordering is covered by tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: daniel-lxs

.changeset/steady-discord-gateway.md

@@ -0,0 +1,11 @@
+---
+'@roomote/web': patch
+---
+
+Harden the Discord gateway: quarantined (undeliverable) events now surface in
+the Discord settings diagnostics instead of accumulating invisibly, the
+durable inbound and dead-letter streams are bounded with capacity pressure
+reported before shedding, a single transient Redis blip no longer drops a
+healthy Gateway connection or ratchets delivery restarts to the maximum
+backoff forever, and a dead gateway supervisor reports to error tracking
+instead of only logging.

apps/bullmq/src/index.ts

@@ -31,7 +31,7 @@ import {
   createAdminDashboardMiddleware,
   resolveAdminDashboardAuth,
 } from './admin-auth';
-import { initBullMqSentry } from './monitoring/sentry';
+import { captureBullMqMessage, initBullMqSentry } from './monitoring/sentry';
 import { getRedis, closeRedis } from './redis';
 import { startScheduler } from './scheduler';
 import { startSandboxOidcRefreshQueue } from './sandbox-oidc-refresh-queue';
@@ -69,12 +69,25 @@ const redis = getRedis();
 
 initBullMqSentry();
 
-const discordGatewaySupervisor = startDiscordGatewaySupervisor(redis, {
-  ...process.env,
-  ENCRYPTION_KEY: Env.ENCRYPTION_KEY,
-  R_DISCORD_GATEWAY_SECRET: Env.R_DISCORD_GATEWAY_SECRET,
-  TRPC_URL: Env.TRPC_URL,
-});
+const discordGatewaySupervisor = startDiscordGatewaySupervisor(
+  redis,
+  {
+    ...process.env,
+    ENCRYPTION_KEY: Env.ENCRYPTION_KEY,
+    R_DISCORD_GATEWAY_SECRET: Env.R_DISCORD_GATEWAY_SECRET,
+    TRPC_URL: Env.TRPC_URL,
+  },
+  {
+    // A dead supervisor silently stops Discord ingestion while this process
+    // stays green; make sure it pages instead of only logging.
+    onFatal: (error) =>
+      captureBullMqMessage(
+        `Discord gateway supervisor stopped unexpectedly: ${error instanceof Error ? error.message : String(error)}`,
+        undefined,
+        { component: 'discord-gateway', signal: 'discord-gateway-fatal' },
+      ),
+  },
+);
 
 const { schedulerQueue, schedulerWorker, schedulerQueueEvents } =
   startScheduler();

apps/discord-gateway/src/config.ts

@@ -4,6 +4,8 @@ const DEFAULT_DELIVERY_MAX_ATTEMPTS = 8;
 const DEFAULT_DELIVERY_MAX_BACKOFF_MS = 30_000;
 const DEFAULT_LOGIN_RETRY_BASE_MS = 15_000;
 const DEFAULT_LOGIN_RETRY_MAX_MS = 5 * 60_000;
+const DEFAULT_INBOUND_MAX_ENTRIES = 50_000;
+const DEFAULT_DEAD_LETTER_MAX_ENTRIES = 1_000;
 
 function positiveInteger(value: string | undefined, fallback: number): number {
   const parsed = Number(value);
@@ -33,6 +35,8 @@ export type DiscordGatewayConfig = {
   deliveryPollMs: number;
   deliveryMaxAttempts: number;
   deliveryMaxBackoffMs: number;
+  inboundMaxEntries: number;
+  deadLetterMaxEntries: number;
   loginRetryBaseMs: number;
   loginRetryMaxMs: number;
 };
@@ -65,6 +69,14 @@ export function resolveDiscordGatewayConfig(
       env.DISCORD_GATEWAY_DELIVERY_MAX_BACKOFF_MS,
       DEFAULT_DELIVERY_MAX_BACKOFF_MS,
     ),
+    inboundMaxEntries: positiveInteger(
+      env.DISCORD_GATEWAY_INBOUND_MAX_ENTRIES,
+      DEFAULT_INBOUND_MAX_ENTRIES,
+    ),
+    deadLetterMaxEntries: positiveInteger(
+      env.DISCORD_GATEWAY_DEAD_LETTER_MAX_ENTRIES,
+      DEFAULT_DEAD_LETTER_MAX_ENTRIES,
+    ),
     loginRetryBaseMs: positiveInteger(
       env.DISCORD_GATEWAY_LOGIN_RETRY_BASE_MS,
       DEFAULT_LOGIN_RETRY_BASE_MS,

apps/discord-gateway/src/delivery-loop.test.ts

@@ -155,6 +155,7 @@ export async function runSupervisedDeliveryLoop(
   let restartDelayMs = params.pollMs;
 
   while (!params.signal.aborted) {
+    const startedAt = Date.now();
     try {
       // This flag represents a running delivery worker, not simply the
       // presence of an API secret. Any unexpected exit clears it first.
@@ -164,6 +165,12 @@ export async function runSupervisedDeliveryLoop(
       throw new Error('Discord delivery loop exited unexpectedly');
     } catch (error) {
       if (params.signal.aborted) break;
+      // A long healthy run means this failure is a fresh incident, not part
+      // of the previous one — start the backoff over instead of ratcheting
+      // toward the lifetime maximum.
+      if (Date.now() - startedAt > restartMaxBackoffMs * 2) {
+        restartDelayMs = params.pollMs;
+      }
       await params.status
         .update({
           forwardingReady: false,

apps/discord-gateway/src/delivery-loop.ts

@@ -15,6 +15,15 @@ export type DiscordGatewaySupervisor = {
 export function startDiscordGatewaySupervisor(
   redis: Redis,
   env: NodeJS.ProcessEnv = process.env,
+  options: {
+    /**
+     * Invoked when the supervisor itself dies. Without it the only signal is
+     * a console line while the host process stays green and Discord
+     * ingestion is silently stopped — hosts should report this to their
+     * error tracker.
+     */
+    onFatal?: (error: unknown) => void;
+  } = {},
 ): DiscordGatewaySupervisor {
   const service = new DiscordGatewayService(
     redis,
@@ -24,6 +33,15 @@ export function startDiscordGatewaySupervisor(
     console.error(
       `[discord-gateway] supervisor stopped unexpectedly: ${error instanceof Error ? error.message : String(error)}`,
     );
+    void service.status
+      .update({
+        phase: 'error',
+        ready: false,
+        connected: false,
+        lastError: `Discord gateway supervisor stopped: ${error instanceof Error ? error.message : String(error)}`,
+      })
+      .catch(() => undefined);
+    options.onFatal?.(error);
   });
 
   return {