From 073954d62e0ecd5abc29ac0cb4c69034d0e91eb0 Mon Sep 17 00:00:00 2001 From: John McLear Date: Sat, 16 May 2026 07:24:05 +0100 Subject: [PATCH 1/3] feat(scaling): engine.io socket flush deferral (#7756 / #7767) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the closed engine.io WS packing prototype (#7772). That patch only modified transport.send(packets[]) and never fired because engine.io's Socket.sendPacket calls flush() synchronously after each push to writeBuffer — and flush drains immediately when transport.writable is true (microseconds on WebSocket). The writeBuffer almost never contained more than one packet. This patch flips that. Socket.prototype.sendPacket is re-implemented to push to writeBuffer and then schedule a single coalesced flush via queueMicrotask. Multiple sendPacket calls in the same task all accumulate; the queued microtask drains the whole batch. The transport.send([packets]) call then sees N > 1 packets in steady state, which is where lever 8 / future engine.io transport packing work has the opportunity to coalesce to one WS frame. Microtask deferral adds zero meaningful wall-clock latency: microtasks drain before the next macrotask, so anything waiting on the next I/O callback / timer still sees the flush completed first. Wire bytes are unchanged. Gated by settings.engineFlushDefer. Default false. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/node/hooks/express/socketio.ts | 8 +++ src/node/utils/EngineFlushDeferral.ts | 95 +++++++++++++++++++++++++++ src/node/utils/Settings.ts | 14 ++++ 3 files changed, 117 insertions(+) create mode 100644 src/node/utils/EngineFlushDeferral.ts diff --git a/src/node/hooks/express/socketio.ts b/src/node/hooks/express/socketio.ts index 79ef892760b..4734ee98dc3 100644 --- a/src/node/hooks/express/socketio.ts +++ b/src/node/hooks/express/socketio.ts @@ -69,6 +69,14 @@ const socketSessionMiddleware = (args: any) => (socket: any, next: Function) => }; export const expressCreateServer = (hookName:string, args:ArgsExpressType, cb:Function) => { + // Engine.io socket flush deferral (#7756 / #7767). Apply BEFORE building + // the socket.io Server so the patched Socket prototype is in effect when + // the Server creates its engine. + if (settings.engineFlushDefer === true) { + // eslint-disable-next-line @typescript-eslint/no-require-imports + require('../../utils/EngineFlushDeferral').installEngineFlushDeferral(); + } + // init socket.io and redirect all requests to the MessageHandler // there shouldn't be a browser that isn't compatible to all // transports in this list at once diff --git a/src/node/utils/EngineFlushDeferral.ts b/src/node/utils/EngineFlushDeferral.ts new file mode 100644 index 00000000000..e8c15b4ea6f --- /dev/null +++ b/src/node/utils/EngineFlushDeferral.ts @@ -0,0 +1,95 @@ +// Engine.io socket flush deferral — #7756 / #7767 deeper investigation +// after the simple WS transport-level packing prototype (#7772) showed +// that the writeBuffer almost never accumulates because flush() drains +// immediately on `transport.writable === true`. +// +// engine.io's Socket.sendPacket(...) ends with: +// +// this.writeBuffer.push(packet); +// if (callback) this.packetsFn.push(callback); +// this.flush(); // <-- synchronous +// +// flush() reads writeBuffer and hands it to transport.send. For +// WebSocket, transport.writable is true again within microseconds of +// each write, so each sendPacket() call drains a buffer of size 1. The +// transport.send([packets]) function then iterates packets and writes +// one WS frame per packet — which is what the polling transport's +// natural encodePayload batching avoids. +// +// This patch coalesces synchronous-task sendPacket calls onto a single +// microtask-scheduled flush. Inside the same JS task, multiple +// sendPacket() calls accumulate in writeBuffer; the queued microtask +// then calls flush() once with the whole batch. The transport's +// send([batch]) sees N > 1 packets and the WS payload-encoding fast +// path (also added by lever 8) coalesces them into one frame. +// +// Microtask deferral adds zero meaningful wall-clock latency: +// microtasks drain before the next macrotask, so any consumer waiting +// on the next setImmediate / setTimeout / I/O callback still sees the +// flush completed. +// +// Forward-compatible. Existing clients receive identical wire bytes +// because the engine.io packet encoding is unchanged; the difference +// is only how many engine.io packets share one transport-level send +// call. The WS transport's send([packets]) path is then where lever 8 +// (or this patch's accompanying engine-packing branch) decides +// whether to ship them as N frames or one payload-encoded frame. +// +// Gated by settings.engineFlushDefer. Default off; production unaffected. + +import log4js from 'log4js'; + +const logger = log4js.getLogger('engine-flush-defer'); + +let installed = false; + +const SCHEDULED = Symbol('engineFlushScheduled'); + +export const installEngineFlushDeferral = (): void => { + if (installed) return; + installed = true; + + let SocketProto: {sendPacket: (...a: unknown[]) => unknown}; + try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + SocketProto = require('engine.io/build/socket').Socket.prototype; + } catch (err: any) { + logger.warn(`Unable to install engine.io flush deferral (module not found): ${err && err.message || err}`); + return; + } + if (typeof SocketProto.sendPacket !== 'function') { + logger.warn('engine.io Socket shape unexpected; skipping flush deferral patch'); + return; + } + + // Re-implementing sendPacket inline rather than wrapping the original + // so the single closing `this.flush()` becomes a microtask-coalesced + // schedule. The body is intentionally a near-verbatim copy of the + // engine.io 6.6.5 implementation so future engine.io upgrades that + // change packet-shape semantics still need re-vetting. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + SocketProto.sendPacket = function (this: any, type: any, data: any, options: any, callback: any) { + if ('function' === typeof options) { + callback = options; + options = {}; + } + if ('closing' === this.readyState || 'closed' === this.readyState) return; + + options = options || {}; + options.compress = options.compress !== false; + const packet: any = {type, options}; + if (data !== undefined) packet.data = data; + this.emit('packetCreate', packet); + this.writeBuffer.push(packet); + if ('function' === typeof callback) this.packetsFn.push(callback); + + if (this[SCHEDULED]) return; + this[SCHEDULED] = true; + queueMicrotask(() => { + this[SCHEDULED] = false; + this.flush(); + }); + }; + + logger.info('engine.io socket flush deferral enabled (#7756 / #7767)'); +}; diff --git a/src/node/utils/Settings.ts b/src/node/utils/Settings.ts index 97413004100..76a687746bd 100644 --- a/src/node/utils/Settings.ts +++ b/src/node/utils/Settings.ts @@ -272,6 +272,7 @@ export type SettingsType = { automaticReconnectionTimeout: number, loadTest: boolean, scalingDiveMetrics: boolean, + engineFlushDefer: boolean, dumpOnUncleanExit: boolean, indentationOnNewLine: boolean, logconfig: any | null, @@ -658,6 +659,19 @@ const settings: SettingsType = { * production deployments aren't paying for instrumentation they don't use. */ scalingDiveMetrics: false, + /** + * Defer engine.io socket flush onto the next microtask so multiple + * sendPacket() calls within the same task accumulate in the writeBuffer + * before drain. Pairs with engine.io's existing transport.send([packets]) + * fast path so a batched send produces fewer WebSocket frames. + * + * Adds no meaningful wall-clock latency — microtasks drain before any + * subsequent macrotask. Backward-compatible at the wire level; existing + * clients receive identical packet bytes. + * + * Default false. Enable only when scoring under the scaling dive. + */ + engineFlushDefer: false, /** * Disable dump of objects preventing a clean exit */ From f0661a9d8e9f6832bf0c9c1bf33be84990e3c607 Mon Sep 17 00:00:00 2001 From: John McLear Date: Sat, 16 May 2026 07:42:34 +0100 Subject: [PATCH 2/3] fix(engine-flush-defer): address Qodo review Two issues from the initial review: 1. Install guard blocked retries. Setting `installed = true` BEFORE requiring + validating engine.io meant a transient require error permanently disabled the patch for the rest of the process. Now the flag is set only after both checks pass; on failure, the warning is logged and a later boot path can retry. 2. engineFlushDefer was undocumented in settings.json.template. Added with the same prose as Settings.ts, including the "wire bytes are unchanged" / "no meaningful wall-clock latency" notes so operators see the safety case. Co-Authored-By: Claude Opus 4.7 (1M context) --- settings.json.template | 13 +++++++++++++ src/node/utils/EngineFlushDeferral.ts | 10 +++++++--- 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/settings.json.template b/settings.json.template index e88e82a36af..c37bfe7f928 100644 --- a/settings.json.template +++ b/settings.json.template @@ -733,6 +733,19 @@ */ "loadTest": false, + /* + * Defer engine.io socket flush onto the next microtask so multiple + * sendPacket() calls in the same task accumulate in writeBuffer before + * the underlying transport.send drains. Pairs with engine.io's existing + * batched-send path so high-fan-out scenarios produce fewer WebSocket + * frames. Microtask deferral adds no meaningful wall-clock latency — + * microtasks drain before any subsequent macrotask. Wire bytes are + * unchanged. + * + * #7756 / #7767. Default off; production unaffected. + */ + "engineFlushDefer": false, + /** * Disable dump of objects preventing a clean exit */ diff --git a/src/node/utils/EngineFlushDeferral.ts b/src/node/utils/EngineFlushDeferral.ts index e8c15b4ea6f..36b3804ed2b 100644 --- a/src/node/utils/EngineFlushDeferral.ts +++ b/src/node/utils/EngineFlushDeferral.ts @@ -47,7 +47,6 @@ const SCHEDULED = Symbol('engineFlushScheduled'); export const installEngineFlushDeferral = (): void => { if (installed) return; - installed = true; let SocketProto: {sendPacket: (...a: unknown[]) => unknown}; try { @@ -55,12 +54,17 @@ export const installEngineFlushDeferral = (): void => { SocketProto = require('engine.io/build/socket').Socket.prototype; } catch (err: any) { logger.warn(`Unable to install engine.io flush deferral (module not found): ${err && err.message || err}`); - return; + return; // Leave `installed` false so a later boot path can retry. } if (typeof SocketProto.sendPacket !== 'function') { logger.warn('engine.io Socket shape unexpected; skipping flush deferral patch'); - return; + return; // Leave `installed` false so a later boot path can retry. } + // Only after both require and shape check succeed do we record that the + // patch is installed. Setting the flag before validation (the original + // code) would have permanently disabled retries after a transient + // require failure in test/CI environments where socket.io may load late. + installed = true; // Re-implementing sendPacket inline rather than wrapping the original // so the single closing `this.flush()` becomes a microtask-coalesced From f5b91431fd16049122f156e4016a819e58777e03 Mon Sep 17 00:00:00 2001 From: SamTV12345 <40429738+samtv12345@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:41:13 +0200 Subject: [PATCH 3/3] fix(scaling): make the flush deferral actually install, and test it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `require('engine.io/build/socket')` cannot resolve, in two independent ways: engine.io is a transitive dependency of socket.io and therefore invisible from `src` under pnpm (MODULE_NOT_FOUND), and `build/socket` is not in engine.io's `exports` map, so the deep path fails with ERR_PACKAGE_PATH_NOT_EXPORTED even where the package is resolvable. The failure was downgraded to a warning, so `engineFlushDefer: true` left the stock synchronous flush in place and the feature was inert — which means the N=3 numbers in the PR description cannot have measured this patch. Resolve engine.io through its public entry point relative to socket.io instead, which also guarantees we patch the copy socket.io actually loaded rather than a second one a hoisting layout might provide. Also sync the copied body to the installed engine.io 6.6.9: upstream attaches the payload on a truthiness check, not `!== undefined`. New spec covers coalescing, packet order, the closing/closed early return, send callbacks, the callback-in-options-slot form, upstream's payload semantics, and pins the SHA of engine.io's own sendPacket so an upgrade that touches it fails here and forces a re-vet of the copy. It restores the prototype afterwards — otherwise every later spec in the suite runs with the feature silently enabled. Co-Authored-By: Claude Opus 5 (1M context) --- src/node/utils/EngineFlushDeferral.ts | 24 ++- .../backend/specs/engineFlushDeferral.ts | 143 ++++++++++++++++++ 2 files changed, 161 insertions(+), 6 deletions(-) create mode 100644 src/tests/backend/specs/engineFlushDeferral.ts diff --git a/src/node/utils/EngineFlushDeferral.ts b/src/node/utils/EngineFlushDeferral.ts index 36b3804ed2b..a9d38af0f3e 100644 --- a/src/node/utils/EngineFlushDeferral.ts +++ b/src/node/utils/EngineFlushDeferral.ts @@ -50,10 +50,19 @@ export const installEngineFlushDeferral = (): void => { let SocketProto: {sendPacket: (...a: unknown[]) => unknown}; try { + // Resolve engine.io through socket.io, and only via its public entry point: + // - engine.io is a transitive dependency of socket.io, so under pnpm's + // layout it is not resolvable from `src` at all (MODULE_NOT_FOUND), and + // - `engine.io/build/socket` is not in engine.io's `exports` map, so a deep + // require fails with ERR_PACKAGE_PATH_NOT_EXPORTED even when it is. + // Resolving relative to socket.io also guarantees we patch the very copy + // socket.io loaded, not a second one that a hoisting layout might provide. // eslint-disable-next-line @typescript-eslint/no-require-imports - SocketProto = require('engine.io/build/socket').Socket.prototype; + const engineIo = require(require.resolve('engine.io', {paths: [require.resolve('socket.io')]})); + SocketProto = engineIo.Socket.prototype; } catch (err: any) { - logger.warn(`Unable to install engine.io flush deferral (module not found): ${err && err.message || err}`); + logger.warn('engineFlushDefer is enabled but the engine.io Socket class could not be ' + + `resolved, so the patch is NOT active: ${err && err.message || err}`); return; // Leave `installed` false so a later boot path can retry. } if (typeof SocketProto.sendPacket !== 'function') { @@ -68,9 +77,10 @@ export const installEngineFlushDeferral = (): void => { // Re-implementing sendPacket inline rather than wrapping the original // so the single closing `this.flush()` becomes a microtask-coalesced - // schedule. The body is intentionally a near-verbatim copy of the - // engine.io 6.6.5 implementation so future engine.io upgrades that - // change packet-shape semantics still need re-vetting. + // schedule. The body is a verbatim copy of engine.io 6.6.9's + // implementation apart from that last line — engineFlushDeferral.ts pins the + // upstream source with a fingerprint, so an engine.io upgrade that touches + // sendPacket fails the test suite and forces a re-vet of this copy. // eslint-disable-next-line @typescript-eslint/no-explicit-any SocketProto.sendPacket = function (this: any, type: any, data: any, options: any, callback: any) { if ('function' === typeof options) { @@ -82,7 +92,9 @@ export const installEngineFlushDeferral = (): void => { options = options || {}; options.compress = options.compress !== false; const packet: any = {type, options}; - if (data !== undefined) packet.data = data; + // Upstream uses a truthiness check, not `!== undefined`: a falsy payload is + // deliberately not attached (encodePacket renders `type` alone for it). + if (data) packet.data = data; this.emit('packetCreate', packet); this.writeBuffer.push(packet); if ('function' === typeof callback) this.packetsFn.push(callback); diff --git a/src/tests/backend/specs/engineFlushDeferral.ts b/src/tests/backend/specs/engineFlushDeferral.ts new file mode 100644 index 00000000000..2c89f623caa --- /dev/null +++ b/src/tests/backend/specs/engineFlushDeferral.ts @@ -0,0 +1,143 @@ +'use strict'; + +/** + * Tests for the opt-in engine.io flush deferral (settings.engineFlushDefer). + * + * Two jobs: + * 1. Prove the patch actually installs and coalesces. The first cut resolved the + * Socket class via `require('engine.io/build/socket')`, which cannot work: + * engine.io is a transitive dependency of socket.io (not resolvable from + * `src` under pnpm) and `build/socket` is not in its `exports` map. The + * failure was caught and logged, so the feature was silently inert. + * 2. Pin engine.io's own sendPacket source. The patch re-implements that method, + * so an engine.io upgrade that changes it must fail here and force a re-vet + * rather than silently diverging. + */ + +const assert = require('assert').strict; +const crypto = require('crypto'); +const {installEngineFlushDeferral} = require('../../../node/utils/EngineFlushDeferral'); + +// sha256 of engine.io 6.6.9's Socket.prototype.sendPacket source. Update this +// ONLY together with a re-read of the upstream method against the copy in +// EngineFlushDeferral.ts. +const PINNED_SEND_PACKET_SHA = + 'a94abb8dd747d4f55bc7611143b185d384463afd5ee9f006a5cde5a1851c0a05'; + +const engineIoSocketProto = () => { + const path = require.resolve('engine.io', {paths: [require.resolve('socket.io')]}); + return require(path).Socket.prototype; +}; + +// Minimal stand-in for an engine.io Socket: the patch only touches these. +const fakeSocket = () => ({ + readyState: 'open', + writeBuffer: [] as any[], + packetsFn: [] as any[], + flushed: 0, + flushedSizes: [] as number[], + emit() {}, + flush() { + this.flushed++; + this.flushedSizes.push(this.writeBuffer.length); + this.writeBuffer = []; + }, +}); + +describe(__filename, function () { + let sendPacket: any; + let upstreamSendPacket: any; + + before(function () { + // Capture upstream BEFORE patching, so the drift check below compares against + // the real engine.io implementation. + upstreamSendPacket = engineIoSocketProto().sendPacket; + installEngineFlushDeferral(); + sendPacket = engineIoSocketProto().sendPacket; + }); + + after(function () { + // The patch mutates a shared prototype, so leaving it in place would silently + // run every later spec in the suite with the feature enabled. + engineIoSocketProto().sendPacket = upstreamSendPacket; + }); + + it('installs onto the engine.io copy socket.io actually loaded', function () { + // The bug this guards: a resolution failure downgraded to a warning, leaving + // the stock synchronous implementation in place. + assert.match(sendPacket.toString(), /queueMicrotask/, + 'engineFlushDefer did not patch engine.io Socket.prototype.sendPacket'); + }); + + it('coalesces packets sent in one task into a single flush', async function () { + const s = fakeSocket(); + sendPacket.call(s, 'message', 'a'); + sendPacket.call(s, 'message', 'b'); + sendPacket.call(s, 'message', 'c'); + assert.equal(s.flushed, 0, 'flush must not happen synchronously'); + assert.equal(s.writeBuffer.length, 3, 'packets accumulate in writeBuffer'); + + await Promise.resolve(); + assert.equal(s.flushed, 1, 'exactly one flush per task'); + assert.deepEqual(s.flushedSizes, [3], 'the flush carries the whole batch'); + }); + + it('preserves packet order and payloads', async function () { + const s = fakeSocket(); + const seen: any[] = []; + s.flush = function () { seen.push(...this.writeBuffer.map((p: any) => p.data)); }; + for (const d of ['1', '2', '3']) sendPacket.call(s, 'message', d); + await Promise.resolve(); + assert.deepEqual(seen, ['1', '2', '3']); + }); + + it('schedules a fresh flush for the next task', async function () { + const s = fakeSocket(); + sendPacket.call(s, 'message', 'a'); + await Promise.resolve(); + sendPacket.call(s, 'message', 'b'); + await Promise.resolve(); + assert.equal(s.flushed, 2); + assert.deepEqual(s.flushedSizes, [1, 1]); + }); + + it('drops packets once the socket is closing or closed', async function () { + for (const readyState of ['closing', 'closed']) { + const s = fakeSocket(); + s.readyState = readyState; + sendPacket.call(s, 'message', 'a'); + await Promise.resolve(); + assert.equal(s.writeBuffer.length, 0, `${readyState}: nothing queued`); + assert.equal(s.flushed, 0, `${readyState}: nothing flushed`); + } + }); + + it('keeps send callbacks, and treats a callback in the options slot as one', async function () { + const s = fakeSocket(); + const cb = () => {}; + sendPacket.call(s, 'message', 'a', cb); + assert.deepEqual(s.packetsFn, [cb], 'callback passed in the options position'); + assert.equal(s.writeBuffer[0].options.compress, true, 'compression defaults to on'); + await Promise.resolve(); + }); + + it('attaches only truthy payloads, like upstream', async function () { + const s = fakeSocket(); + sendPacket.call(s, 'pong'); + sendPacket.call(s, 'message', ''); + assert.equal('data' in s.writeBuffer[0], false, 'no data for a bare pong'); + assert.equal('data' in s.writeBuffer[1], false, 'no data for an empty payload'); + await Promise.resolve(); + }); + + it('engine.io sendPacket is unchanged since the copy was vetted', function () { + assert.doesNotMatch(upstreamSendPacket.toString(), /queueMicrotask/, + 'the patch was already installed before this spec ran, so upstream could not ' + + 'be captured — run this spec in its own process to check for drift'); + const sha = crypto.createHash('sha256').update(upstreamSendPacket.toString()).digest('hex'); + assert.equal(sha, PINNED_SEND_PACKET_SHA, + 'engine.io Socket.prototype.sendPacket changed upstream — re-vet the copy in ' + + 'EngineFlushDeferral.ts against the new implementation, then update ' + + `PINNED_SEND_PACKET_SHA to ${sha}`); + }); +});