fix(gateway): reject slow node event sends

samzong · samzong · commit 664e05e50d76 · 2026-05-20T09:20:46.000+08:00
Signed-off-by: samzong &lt;samzong.lu@gmail.com&gt;
diff --git a/src/gateway/node-registry.test.ts b/src/gateway/node-registry.test.ts
@@ -1,6 +1,7 @@
 import { EventEmitter } from "node:events";
 import { describe, expect, it, vi } from "vitest";
 import { NodeRegistry, serializeEventPayload } from "./node-registry.js";
+import { MAX_BUFFERED_BYTES } from "./server-constants.js";
 import type { GatewayWsClient } from "./server/ws-types.js";
 
 function makeClient(
@@ -506,6 +507,26 @@ describe("gateway/node-registry", () => {
     ]);
   });
 
+  it("rejects raw event sends when the node socket buffer is saturated", () => {
+    const registry = new NodeRegistry();
+    const socket = {
+      bufferedAmount: MAX_BUFFERED_BYTES + 1,
+      send: vi.fn(),
+      close: vi.fn(),
+    };
+    registry.register(
+      makeClient("conn-1", "node-1", [], {
+        socket: socket as unknown as GatewayWsClient["socket"],
+      }),
+      {},
+    );
+    const payload = serializeEventPayload({ foo: "bar" });
+
+    expect(registry.sendEventRaw("node-1", "chat", payload)).toBe(false);
+    expect(socket.send).not.toHaveBeenCalled();
+    expect(socket.close).toHaveBeenCalledWith(1008, "slow consumer");
+  });
+
   it("refreshes effective live surface within the declared surface", () => {
     const registry = new NodeRegistry();
     const client = makeClient("conn-1", "node-1", [], {
diff --git a/src/gateway/node-registry.ts b/src/gateway/node-registry.ts
@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import { MAX_BUFFERED_BYTES } from "./server-constants.js";
 import type { GatewayWsClient } from "./server/ws-types.js";
 
 export type NodeSession = {
@@ -72,6 +73,7 @@ type PingableSocket = {
 const SERIALIZED_EVENT_PAYLOAD = Symbol("openclaw.serializedEventPayload");
 const AUTHORIZED_SYSTEM_RUN_EVENT_GRACE_MS = 5 * 60 * 1000;
 const WEBSOCKET_OPEN_READY_STATE = 1;
+const SLOW_CONSUMER_CLOSE_CODE = 1008;
 
 export type SerializedEventPayload = {
   readonly json: string;
@@ -657,6 +659,9 @@ export class NodeRegistry {
   }
 
   private sendEventInternal(node: NodeSession, event: string, payload: unknown): boolean {
+    if (this.rejectSlowNodeSocket(node)) {
+      return false;
+    }
     try {
       node.client.socket.send(
         JSON.stringify({
@@ -683,6 +688,9 @@ export class NodeRegistry {
     ) {
       return false;
     }
+    if (this.rejectSlowNodeSocket(node)) {
+      return false;
+    }
     try {
       const payloadFragment = payloadJSON ? `,"payload":${payloadJSON.json}` : "";
       node.client.socket.send(
@@ -697,4 +705,16 @@ export class NodeRegistry {
   private sendEventToSession(node: NodeSession, event: string, payload: unknown): boolean {
     return this.sendEventInternal(node, event, payload);
   }
+
+  private rejectSlowNodeSocket(node: NodeSession): boolean {
+    if (!(node.client.socket.bufferedAmount > MAX_BUFFERED_BYTES)) {
+      return false;
+    }
+    try {
+      node.client.socket.close(SLOW_CONSUMER_CLOSE_CODE, "slow consumer");
+    } catch {
+      /* ignore */
+    }
+    return true;
+  }
 }
