From 69aeba9d862c1a5b79c70af37803fc68b4d8a34e Mon Sep 17 00:00:00 2001 From: xingzhou Date: Sun, 19 Jul 2026 10:14:16 +0800 Subject: [PATCH] fix(discord): sustained gateway bursts stop growing memory (#110954) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * fix(discord): surface gateway overflow warnings Co-authored-by: 张贵萍0668001030 --------- Co-authored-by: Peter Steinberger --- .../discord/src/gateway-logging.test.ts | 7 +++- extensions/discord/src/gateway-logging.ts | 6 ++- .../src/internal/gateway-rate-limit.ts | 36 +++++++++++++++++ .../discord/src/internal/gateway.test.ts | 39 ++++++++++++++++++- extensions/discord/src/internal/gateway.ts | 5 +++ 5 files changed, 88 insertions(+), 5 deletions(-) diff --git a/extensions/discord/src/gateway-logging.test.ts b/extensions/discord/src/gateway-logging.test.ts index c03c1c9013b5..27bddf546870 100644 --- a/extensions/discord/src/gateway-logging.test.ts +++ b/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(); }); diff --git a/extensions/discord/src/gateway-logging.ts b/extensions/discord/src/gateway-logging.ts index b4df4bc48a6d..129f319ca295 100644 --- a/extensions/discord/src/gateway-logging.ts +++ b/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; @@ -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) => { diff --git a/extensions/discord/src/internal/gateway-rate-limit.ts b/extensions/discord/src/internal/gateway-rate-limit.ts index dfebb1c0181e..258481507d3e 100644 --- a/extensions/discord/src/internal/gateway-rate-limit.ts +++ b/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(); } } diff --git a/extensions/discord/src/internal/gateway.test.ts b/extensions/discord/src/internal/gateway.test.ts index 1d736fb67d55..51fdddfac841 100644 --- a/extensions/discord/src/internal/gateway.test.ts +++ b/extensions/discord/src/internal/gateway.test.ts @@ -50,11 +50,12 @@ function firstSentGatewayPayload(send: ReturnType): 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, }); }); diff --git a/extensions/discord/src/internal/gateway.ts b/extensions/discord/src/internal/gateway.ts index d9c5793dff5d..d28cdcb16057 100644 --- a/extensions/discord/src/internal/gateway.ts +++ b/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) {