Discord: harden gateway resilience

src/discord/monitor.gateway.test.ts

@@ -47,6 +47,31 @@ describe("waitForDiscordGatewayStop", () => {
     expect(disconnect).toHaveBeenCalledTimes(1);
   });
 
+  it("ignores gateway errors when instructed", async () => {
+    const emitter = new EventEmitter();
+    const disconnect = vi.fn();
+    const onGatewayError = vi.fn();
+    const abort = new AbortController();
+    const err = new Error("transient");
+
+    const promise = waitForDiscordGatewayStop({
+      gateway: { emitter, disconnect },
+      abortSignal: abort.signal,
+      onGatewayError,
+      shouldStopOnError: () => false,
+    });
+
+    emitter.emit("error", err);
+    expect(onGatewayError).toHaveBeenCalledWith(err);
+    expect(disconnect).toHaveBeenCalledTimes(0);
+    expect(emitter.listenerCount("error")).toBe(1);
+
+    abort.abort();
+    await expect(promise).resolves.toBeUndefined();
+    expect(disconnect).toHaveBeenCalledTimes(1);
+    expect(emitter.listenerCount("error")).toBe(0);
+  });
+
   it("resolves on abort without a gateway", async () => {
     const abort = new AbortController();
 

src/discord/monitor.gateway.ts

@@ -15,8 +15,9 @@ export async function waitForDiscordGatewayStop(params: {
   gateway?: DiscordGatewayHandle;
   abortSignal?: AbortSignal;
   onGatewayError?: (err: unknown) => void;
+  shouldStopOnError?: (err: unknown) => boolean;
 }): Promise<void> {
-  const { gateway, abortSignal, onGatewayError } = params;
+  const { gateway, abortSignal, onGatewayError, shouldStopOnError } = params;
   const emitter = gateway?.emitter;
   return await new Promise<void>((resolve, reject) => {
     let settled = false;
@@ -49,7 +50,10 @@ export async function waitForDiscordGatewayStop(params: {
     };
     const onGatewayErrorEvent = (err: unknown) => {
       onGatewayError?.(err);
-      finishReject(err);
+      const shouldStop = shouldStopOnError?.(err) ?? true;
+      if (shouldStop) {
+        finishReject(err);
+      }
     };
 
     if (abortSignal?.aborted) {

src/discord/monitor.ts

@@ -324,6 +324,9 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
       publicKey: "a",
       token,
       autoDeploy: nativeEnabled,
+      eventQueue: {
+        listenerTimeout: 120_000,
+      },
     },
     {
       commands,
@@ -331,6 +334,9 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
     },
     [
       new GatewayPlugin({
+        reconnect: {
+          maxAttempts: Number.POSITIVE_INFINITY,
+        },
         intents:
           GatewayIntents.Guilds |
           GatewayIntents.GuildMessages |
@@ -409,18 +415,55 @@ export async function monitorDiscordProvider(opts: MonitorDiscordOpts = {}) {
 
   const gateway = client.getPlugin<GatewayPlugin>("gateway");
   const gatewayEmitter = getDiscordGatewayEmitter(gateway);
-  await waitForDiscordGatewayStop({
-    gateway: gateway
-      ? {
-          emitter: gatewayEmitter,
-          disconnect: () => gateway.disconnect(),
-        }
-      : undefined,
-    abortSignal: opts.abortSignal,
-    onGatewayError: (err) => {
-      runtime.error?.(danger(`discord gateway error: ${String(err)}`));
-    },
-  });
+  let queueMetricsTimer: ReturnType<typeof setInterval> | undefined;
+  let lastQueueTimeouts = 0;
+  let lastQueueDropped = 0;
+  const onGatewayWarning = (warning: unknown) => {
+    logVerbose(`discord gateway warning: ${String(warning)}`);
+  };
+  if (shouldLogVerbose()) {
+    gatewayEmitter?.on("warning", onGatewayWarning);
+    queueMetricsTimer = setInterval(() => {
+      const metrics = client.eventHandler?.getMetrics?.();
+      if (!metrics) return;
+      const nearCapacity =
+        metrics.maxQueueSize > 0 &&
+        metrics.queueSize / metrics.maxQueueSize >= 0.8;
+      const hasNewTimeouts = metrics.timeouts > lastQueueTimeouts;
+      const hasNewDrops = metrics.dropped > lastQueueDropped;
+      if (nearCapacity || hasNewTimeouts || hasNewDrops) {
+        logVerbose(`discord event queue metrics: ${JSON.stringify(metrics)}`);
+      }
+      lastQueueTimeouts = metrics.timeouts;
+      lastQueueDropped = metrics.dropped;
+    }, 60000);
+  }
+  try {
+    await waitForDiscordGatewayStop({
+      gateway: gateway
+        ? {
+            emitter: gatewayEmitter,
+            disconnect: () => gateway.disconnect(),
+          }
+        : undefined,
+      abortSignal: opts.abortSignal,
+      onGatewayError: (err) => {
+        runtime.error?.(danger(`discord gateway error: ${String(err)}`));
+      },
+      shouldStopOnError: (err) => {
+        const message = String(err);
+        return (
+          message.includes("Max reconnect attempts") ||
+          message.includes("Fatal Gateway error")
+        );
+      },
+    });
+  } finally {
+    if (queueMetricsTimer) {
+      clearInterval(queueMetricsTimer);
+    }
+    gatewayEmitter?.removeListener("warning", onGatewayWarning);
+  }
 }
 
 async function clearDiscordNativeCommands(params: {