feat(scaling): engine.io socket flush deferral — modest tail-latency reduction at mid-load

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
   */

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

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)');
+};

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
    */

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}`);
+  });
+});