diff --git a/settings.json.template b/settings.json.template index 5c01460a60c..d32b5537e21 100644 --- a/settings.json.template +++ b/settings.json.template @@ -789,6 +789,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/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..a9d38af0f3e --- /dev/null +++ b/src/node/utils/EngineFlushDeferral.ts @@ -0,0 +1,111 @@ +// 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; + + 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 + const engineIo = require(require.resolve('engine.io', {paths: [require.resolve('socket.io')]})); + SocketProto = engineIo.Socket.prototype; + } catch (err: any) { + 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') { + logger.warn('engine.io Socket shape unexpected; skipping flush deferral patch'); + 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 + // 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) { + callback = options; + options = {}; + } + if ('closing' === this.readyState || 'closed' === this.readyState) return; + + options = options || {}; + options.compress = options.compress !== false; + const packet: any = {type, options}; + // 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); + + 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 f1ed4e4e085..499e35b0531 100644 --- a/src/node/utils/Settings.ts +++ b/src/node/utils/Settings.ts @@ -285,6 +285,7 @@ export type SettingsType = { automaticReconnectionTimeout: number, loadTest: boolean, scalingDiveMetrics: boolean, + engineFlushDefer: boolean, dumpOnUncleanExit: boolean, indentationOnNewLine: boolean, logconfig: any | null, @@ -718,6 +719,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 */ 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}`); + }); +});