[Fix] Reject slow node event sends (#84387)

Merged via squash.

Prepared head SHA: b459f9ea57
Co-authored-by: samzong <13782141+samzong@users.noreply.github.com>
Co-authored-by: frankekn <4488090+frankekn@users.noreply.github.com>
Reviewed-by: @frankekn

CHANGELOG.md

@@ -91,6 +91,7 @@ Docs: https://docs.openclaw.ai
 - CLI/gateway: include the running Gateway version in `gateway status` JSON output, preserving existing server metadata while falling back to status RPC data for read probes. Fixes #56222. Thanks @galiniliev.
 - Memory/search: close local embedding providers when active-memory searches time out so pending local model loads and embedding contexts are aborted and released. (#83858) Thanks @brokemac79.
 - CLI/nodes: request pending node surface approval scopes before `openclaw nodes approve` so exec-capable node approval can use admin-scoped Gateway credentials instead of failing with `missing scope: operator.admin`. (#84392) Thanks @joshavant.
+- Gateway: reject slow node event sends before outbound buffers grow unbounded and log the rejected payload diagnostic. (#84387) Thanks @samzong.
 - Agents: include bounded trajectory queued-writer diagnostics in `pi-trajectory-flush` timeout warnings so flush stalls show pending writes, queued bytes, and append state. Fixes #82961. (#82962) Thanks @galiniliev.
 - Agents/subagents: recover stale completion announces by retrying unsupported transcript-wait wakes without transcript waiting and forcing a message-tool handoff when the requester run is already stale. Fixes #83699. (#83700) Thanks @galiniliev.
 - Agents/subagents: constrain wildcard subagent target allowlists to configured agents while preserving explicitly listed compatibility targets. Fixes #84040. (#84357) Thanks @joshavant.

src/gateway/node-registry.test.ts

@@ -1,6 +1,8 @@
 import { EventEmitter } from "node:events";
 import { describe, expect, it, vi } from "vitest";
+import { onDiagnosticEvent, resetDiagnosticEventsForTest } from "../infra/diagnostic-events.js";
 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 +508,46 @@ describe("gateway/node-registry", () => {
     ]);
   });
 
+  it("rejects raw event sends when the node socket buffer is saturated", () => {
+    resetDiagnosticEventsForTest();
+    const diagnosticEvents: unknown[] = [];
+    const stopDiagnostics = onDiagnosticEvent((event) => diagnosticEvents.push(event));
+    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" });
+
+    try {
+      expect(registry.sendEventRaw("node-1", "chat", payload)).toBe(false);
+      expect(socket.send).not.toHaveBeenCalled();
+      expect(socket.close).toHaveBeenCalledWith(1008, "slow consumer");
+      expect(diagnosticEvents).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            type: "payload.large",
+            action: "rejected",
+            surface: "gateway.ws.outbound_buffer",
+            bytes: MAX_BUFFERED_BYTES + 1,
+            limitBytes: MAX_BUFFERED_BYTES,
+            reason: "ws_send_buffer_close",
+          }),
+        ]),
+      );
+    } finally {
+      stopDiagnostics();
+      resetDiagnosticEventsForTest();
+    }
+  });
+
   it("refreshes effective live surface within the declared surface", () => {
     const registry = new NodeRegistry();
     const client = makeClient("conn-1", "node-1", [], {

src/gateway/node-registry.ts

@@ -1,4 +1,6 @@
 import { randomUUID } from "node:crypto";
+import { logRejectedLargePayload } from "../logging/diagnostic-payload.js";
+import { MAX_BUFFERED_BYTES } from "./server-constants.js";
 import type { GatewayWsClient } from "./server/ws-types.js";
 
 export type NodeSession = {
@@ -72,6 +74,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 +660,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 +689,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 +706,22 @@ 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;
+    }
+    logRejectedLargePayload({
+      surface: "gateway.ws.outbound_buffer",
+      bytes: node.client.socket.bufferedAmount,
+      limitBytes: MAX_BUFFERED_BYTES,
+      reason: "ws_send_buffer_close",
+    });
+    try {
+      node.client.socket.close(SLOW_CONSUMER_CLOSE_CODE, "slow consumer");
+    } catch {
+      /* ignore */
+    }
+    return true;
+  }
 }