fix(discord): sustained gateway bursts stop growing memory (#110954)

* fix(discord): sustained gateway bursts stop growing memory

* fix(discord): contain gateway queue overflow

* fix(discord): drop oldest saturated gateway sends

Co-authored-by: 张贵萍0668001030 <zhang.guiping@xydigit.com>

* fix(discord): surface gateway overflow warnings

Co-authored-by: 张贵萍0668001030 <zhang.guiping@xydigit.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/discord/src/gateway-logging.test.ts

@@ -4,6 +4,7 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 
 vi.mock("openclaw/plugin-sdk/runtime-env", () => ({
   logVerbose: vi.fn(),
+  warn: (message: string) => `warn:${message}`,
 }));
 
 let logVerbose: typeof import("openclaw/plugin-sdk/runtime-env").logVerbose;
@@ -58,7 +59,7 @@ describe("attachDiscordGatewayLogging", () => {
     cleanup();
   });
 
-  it("logs warnings and metrics only to verbose", () => {
+  it("promotes warnings while keeping metrics verbose-only", () => {
     const emitter = new EventEmitter();
     const runtime = makeRuntime();
 
@@ -72,7 +73,9 @@ describe("attachDiscordGatewayLogging", () => {
 
     const logVerboseMock = vi.mocked(logVerbose);
     expect(logVerboseMock).toHaveBeenCalledTimes(2);
-    expect(runtime.log).not.toHaveBeenCalled();
+    expect(runtime.log).toHaveBeenCalledWith(
+      "warn:discord gateway warning: High latency detected: 1200ms",
+    );
 
     cleanup();
   });

extensions/discord/src/gateway-logging.ts

@@ -1,6 +1,6 @@
 // Discord plugin module implements gateway logging behavior.
 import type { EventEmitter } from "node:events";
-import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
+import { logVerbose, warn } from "openclaw/plugin-sdk/runtime-env";
 import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
 
 type GatewayEmitter = Pick<EventEmitter, "on" | "removeListener">;
@@ -49,7 +49,9 @@ export function attachDiscordGatewayLogging(params: {
   };
 
   const onGatewayWarning = (warning: unknown) => {
-    logVerbose(`discord gateway warning: ${String(warning)}`);
+    const message = `discord gateway warning: ${String(warning)}`;
+    logVerbose(message);
+    runtime.log?.(warn(message));
   };
 
   const onGatewayMetrics = (metrics: unknown) => {

extensions/discord/src/internal/gateway-rate-limit.ts

@@ -1,19 +1,30 @@
 // Discord plugin module implements gateway rate limit behavior.
 const GATEWAY_SEND_LIMIT = 120;
 const GATEWAY_SEND_WINDOW_MS = 60_000;
+const GATEWAY_SEND_QUEUE_LIMIT = GATEWAY_SEND_LIMIT;
 
 type QueuedGatewaySend = {
   payload: string;
 };
 
+type GatewaySendQueueOverflow = {
+  droppedEvents: number;
+  maxQueuedEvents: number;
+  policy: "drop-oldest";
+  queuedEvents: number;
+};
+
 export class GatewaySendLimiter {
   private outboundSendTimestamps: number[] = [];
   private outboundQueue: QueuedGatewaySend[] = [];
   private outboundFlushTimer?: NodeJS.Timeout;
+  private droppedEvents = 0;
+  private overflowWarningEmitted = false;
 
   constructor(
     private sendNow: (payload: string) => void,
     private emitError: (error: Error) => void,
+    private emitOverflowWarning: (warning: GatewaySendQueueOverflow) => void,
   ) {}
 
   send(serialized: string, options?: { critical?: boolean }): void {
@@ -21,7 +32,26 @@ export class GatewaySendLimiter {
       this.sendSerialized(serialized);
       return;
     }
+    let shouldEmitOverflowWarning = false;
+    // Keep the newest deferred state within one additional send window. Warn once per
+    // saturation episode so an overload cannot turn into a synchronous log flood.
+    if (this.outboundQueue.length >= GATEWAY_SEND_QUEUE_LIMIT) {
+      this.outboundQueue.shift();
+      this.droppedEvents += 1;
+      if (!this.overflowWarningEmitted) {
+        this.overflowWarningEmitted = true;
+        shouldEmitOverflowWarning = true;
+      }
+    }
     this.outboundQueue.push({ payload: serialized });
+    if (shouldEmitOverflowWarning) {
+      this.emitOverflowWarning({
+        droppedEvents: this.droppedEvents,
+        maxQueuedEvents: GATEWAY_SEND_QUEUE_LIMIT,
+        policy: "drop-oldest",
+        queuedEvents: this.outboundQueue.length,
+      });
+    }
     this.scheduleFlush();
   }
 
@@ -31,6 +61,8 @@ export class GatewaySendLimiter {
       this.outboundFlushTimer = undefined;
     }
     this.outboundQueue = [];
+    this.droppedEvents = 0;
+    this.overflowWarningEmitted = false;
   }
 
   getStatus() {
@@ -45,6 +77,7 @@ export class GatewaySendLimiter {
           : now + GATEWAY_SEND_WINDOW_MS,
       currentEventCount: this.outboundSendTimestamps.length,
       queuedEvents: this.outboundQueue.length,
+      droppedEvents: this.droppedEvents,
     };
   }
 
@@ -100,6 +133,9 @@ export class GatewaySendLimiter {
         return;
       }
     }
+    if (this.outboundQueue.length < GATEWAY_SEND_QUEUE_LIMIT) {
+      this.overflowWarningEmitted = false;
+    }
     this.scheduleFlush();
   }
 }

extensions/discord/src/internal/gateway.test.ts

@@ -50,11 +50,12 @@ function firstSentGatewayPayload(send: ReturnType<typeof attachOpenSocket>): unk
 
 function presenceUpdate(
   status: PresenceUpdateStatus.Online | PresenceUpdateStatus.Idle = PresenceUpdateStatus.Online,
+  since: number | null = null,
 ): GatewaySendPayload {
   return {
     op: GatewayOpcodes.PresenceUpdate,
     d: {
-      since: null,
+      since,
       activities: [],
       status,
       afk: false,
@@ -495,6 +496,7 @@ describe("GatewayPlugin", () => {
       resetTime: 60_000,
       currentEventCount: 120,
       queuedEvents: 1,
+      droppedEvents: 0,
     });
 
     vi.advanceTimersByTime(59_999);
@@ -507,9 +509,43 @@ describe("GatewayPlugin", () => {
       resetTime: 120_000,
       currentEventCount: 1,
       queuedEvents: 0,
+      droppedEvents: 0,
     });
   });
 
+  it("drops the oldest queued events and warns once per saturation episode", () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(0);
+    const gateway = new GatewayPlugin({ autoInteractions: false });
+    const send = attachOpenSocket(gateway);
+    const warningSpy = vi.fn();
+    gateway.emitter.on("warning", warningSpy);
+
+    for (let index = 0; index < 242; index += 1) {
+      gateway.send(presenceUpdate(PresenceUpdateStatus.Online, index));
+    }
+
+    expect(gateway.getRateLimitStatus()).toEqual({
+      remainingEvents: 0,
+      resetTime: 60_000,
+      currentEventCount: 120,
+      queuedEvents: 120,
+      droppedEvents: 2,
+    });
+    expect(warningSpy).toHaveBeenCalledTimes(1);
+    expect(warningSpy).toHaveBeenCalledWith(
+      "Gateway outbound queue overflow policy=drop-oldest droppedEvents=1 queuedEvents=120 maxQueuedEvents=120",
+    );
+
+    vi.advanceTimersByTime(60_000);
+
+    const flushedSinceValues = send.mock.calls.slice(120).map(([serialized]) => {
+      const payload = JSON.parse(String(serialized)) as { d?: { since?: number } };
+      return payload.d?.since;
+    });
+    expect(flushedSinceValues).toEqual(Array.from({ length: 120 }, (_, index) => index + 122));
+  });
+
   it("sends critical gateway events immediately even when regular sends are queued", () => {
     vi.useFakeTimers();
     vi.setSystemTime(0);
@@ -532,6 +568,7 @@ describe("GatewayPlugin", () => {
       resetTime: 60_000,
       currentEventCount: 121,
       queuedEvents: 1,
+      droppedEvents: 0,
     });
   });
 

extensions/discord/src/internal/gateway.ts

@@ -96,6 +96,11 @@ export class GatewayPlugin extends Plugin {
   private outboundLimiter = new GatewaySendLimiter(
     (payload) => this.sendSerializedGatewayEvent(payload),
     (error) => this.emitter.emit("error", error),
+    (warning) =>
+      this.emitter.emit(
+        "warning",
+        `Gateway outbound queue overflow policy=${warning.policy} droppedEvents=${warning.droppedEvents} queuedEvents=${warning.queuedEvents} maxQueuedEvents=${warning.maxQueuedEvents}`,
+      ),
   );
 
   constructor(options: GatewayPluginOptions, gatewayInfo?: APIGatewayBotInfo) {