diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bb188919237..eeebea5ee953 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/gateway/node-registry.test.ts b/src/gateway/node-registry.test.ts index 6d27ceeddb38..722dca8aa005 100644 --- a/src/gateway/node-registry.test.ts +++ b/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", [], { diff --git a/src/gateway/node-registry.ts b/src/gateway/node-registry.ts index d52a3a0283e1..2d523d54f205 100644 --- a/src/gateway/node-registry.ts +++ b/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; + } }