fix: isolate global dispatcher v2 and add Dispatcher1Wrapper bridge (#4827)

Author: mcollina

README.md

@@ -563,6 +563,12 @@ See [Dispatcher.upgrade](./docs/docs/api/Dispatcher.md#dispatcherupgradeoptions-
 Sets the global dispatcher used by Common API Methods. Global dispatcher is shared among compatible undici modules,
 including undici that is bundled internally with node.js.
 
+Undici stores this dispatcher under `Symbol.for('undici.globalDispatcher.2')`.
+
+On Node.js 22, `setGlobalDispatcher()` also mirrors the configured dispatcher to
+`Symbol.for('undici.globalDispatcher.1')` using `Dispatcher1Wrapper`, so Node.js built-in `fetch`
+can keep using the legacy handler contract while Undici uses the new handler API.
+
 ### `undici.getGlobalDispatcher()`
 
 Gets the global dispatcher used by Common API Methods.

docs/docs/api/Dispatcher.md

@@ -232,6 +232,21 @@ Pause/resume now uses the controller:
 
 - Call `controller.pause()` and `controller.resume()` instead of returning `false` from handlers.
 
+#### Compatibility notes
+
+Undici now stores the global dispatcher under `Symbol.for('undici.globalDispatcher.2')`.
+This avoids conflicts with runtimes (such as Node.js built-in `fetch`) that still rely on the legacy dispatcher handler interface.
+
+On Node.js 22, `setGlobalDispatcher()` also mirrors the configured dispatcher to `Symbol.for('undici.globalDispatcher.1')` using a `Dispatcher1Wrapper`, so Node's built-in `fetch` can keep using the legacy handler contract.
+
+If you need to expose a new dispatcher/agent to legacy v1 handler consumers (`onConnect/onHeaders/onData/onComplete/onError/onUpgrade`), use `Dispatcher1Wrapper`:
+
+```js
+import { Agent, Dispatcher1Wrapper } from 'undici'
+
+const legacyCompatibleDispatcher = new Dispatcher1Wrapper(new Agent())
+```
+
 #### Example 1 - Dispatch GET request
 
 ```js

index.js

@@ -6,6 +6,7 @@ const Pool = require('./lib/dispatcher/pool')
 const BalancedPool = require('./lib/dispatcher/balanced-pool')
 const RoundRobinPool = require('./lib/dispatcher/round-robin-pool')
 const Agent = require('./lib/dispatcher/agent')
+const Dispatcher1Wrapper = require('./lib/dispatcher/dispatcher1-wrapper')
 const ProxyAgent = require('./lib/dispatcher/proxy-agent')
 const EnvHttpProxyAgent = require('./lib/dispatcher/env-http-proxy-agent')
 const RetryAgent = require('./lib/dispatcher/retry-agent')
@@ -34,6 +35,7 @@ module.exports.Pool = Pool
 module.exports.BalancedPool = BalancedPool
 module.exports.RoundRobinPool = RoundRobinPool
 module.exports.Agent = Agent
+module.exports.Dispatcher1Wrapper = Dispatcher1Wrapper
 module.exports.ProxyAgent = ProxyAgent
 module.exports.EnvHttpProxyAgent = EnvHttpProxyAgent
 module.exports.RetryAgent = RetryAgent

lib/dispatcher/dispatcher1-wrapper.js

@@ -0,0 +1,101 @@
+'use strict'
+
+const Dispatcher = require('./dispatcher')
+const { InvalidArgumentError } = require('../core/errors')
+const { toRawHeaders } = require('../core/util')
+
+class LegacyHandlerWrapper {
+  #handler
+
+  constructor (handler) {
+    this.#handler = handler
+  }
+
+  onRequestStart (controller, context) {
+    this.#handler.onConnect?.((reason) => controller.abort(reason), context)
+  }
+
+  onRequestUpgrade (controller, statusCode, headers, socket) {
+    const rawHeaders = controller?.rawHeaders ?? toRawHeaders(headers ?? {})
+    this.#handler.onUpgrade?.(statusCode, rawHeaders, socket)
+  }
+
+  onResponseStart (controller, statusCode, headers, statusMessage) {
+    const rawHeaders = controller?.rawHeaders ?? toRawHeaders(headers ?? {})
+
+    if (this.#handler.onHeaders?.(statusCode, rawHeaders, () => controller.resume(), statusMessage) === false) {
+      controller.pause()
+    }
+  }
+
+  onResponseData (controller, chunk) {
+    if (this.#handler.onData?.(chunk) === false) {
+      controller.pause()
+    }
+  }
+
+  onResponseEnd (controller, trailers) {
+    const rawTrailers = controller?.rawTrailers ?? toRawHeaders(trailers ?? {})
+    this.#handler.onComplete?.(rawTrailers)
+  }
+
+  onResponseError (_controller, err) {
+    if (!this.#handler.onError) {
+      throw err
+    }
+
+    this.#handler.onError(err)
+  }
+
+  onBodySent (chunk) {
+    this.#handler.onBodySent?.(chunk)
+  }
+
+  onRequestSent () {
+    this.#handler.onRequestSent?.()
+  }
+
+  onResponseStarted () {
+    this.#handler.onResponseStarted?.()
+  }
+}
+
+class Dispatcher1Wrapper extends Dispatcher {
+  #dispatcher
+
+  constructor (dispatcher) {
+    super()
+
+    if (!dispatcher || typeof dispatcher.dispatch !== 'function') {
+      throw new InvalidArgumentError('Argument dispatcher must implement dispatch')
+    }
+
+    this.#dispatcher = dispatcher
+  }
+
+  static wrapHandler (handler) {
+    if (!handler || typeof handler !== 'object') {
+      throw new InvalidArgumentError('handler must be an object')
+    }
+
+    if (typeof handler.onRequestStart === 'function') {
+      return handler
+    }
+
+    return new LegacyHandlerWrapper(handler)
+  }
+
+  dispatch (opts, handler) {
+    return this.#dispatcher.dispatch(opts, Dispatcher1Wrapper.wrapHandler(handler))
+  }
+
+  close (...args) {
+    return this.#dispatcher.close(...args)
+  }
+
+  destroy (...args) {
+    return this.#dispatcher.destroy(...args)
+  }
+}
+
+module.exports = Dispatcher1Wrapper

lib/global.js

@@ -2,9 +2,13 @@
 
 // We include a version number for the Dispatcher API. In case of breaking changes,
 // this version number must be increased to avoid conflicts.
-const globalDispatcher = Symbol.for('undici.globalDispatcher.1')
+const globalDispatcher = Symbol.for('undici.globalDispatcher.2')
+const legacyGlobalDispatcher = Symbol.for('undici.globalDispatcher.1')
 const { InvalidArgumentError } = require('./core/errors')
 const Agent = require('./dispatcher/agent')
+const Dispatcher1Wrapper = require('./dispatcher/dispatcher1-wrapper')
+
+const nodeMajor = Number(process.versions.node.split('.', 1)[0])
 
 if (getGlobalDispatcher() === undefined) {
   setGlobalDispatcher(new Agent())
@@ -14,12 +18,24 @@ function setGlobalDispatcher (agent) {
   if (!agent || typeof agent.dispatch !== 'function') {
     throw new InvalidArgumentError('Argument agent must implement Agent')
   }
+
   Object.defineProperty(globalThis, globalDispatcher, {
     value: agent,
     writable: true,
     enumerable: false,
     configurable: false
   })
+
+  if (nodeMajor === 22) {
+    const legacyAgent = agent instanceof Dispatcher1Wrapper ? agent : new Dispatcher1Wrapper(agent)
+
+    Object.defineProperty(globalThis, legacyGlobalDispatcher, {
+      value: legacyAgent,
+      writable: true,
+      enumerable: false,
+      configurable: false
+    })
+  }
 }
 
 function getGlobalDispatcher () {

test/node-test/global-dispatcher-version.js

@@ -0,0 +1,100 @@
+'use strict'
+
+const assert = require('node:assert')
+const { test } = require('node:test')
+const { spawnSync } = require('node:child_process')
+const { join } = require('node:path')
+
+const cwd = join(__dirname, '../..')
+
+function runNode (source) {
+  return spawnSync(process.execPath, ['-e', source], {
+    cwd,
+    encoding: 'utf8'
+  })
+}
+
+test('setGlobalDispatcher does not break Node.js global fetch', () => {
+  const script = `
+    const { Agent, setGlobalDispatcher } = require('./index.js')
+    const http = require('node:http')
+    const { once } = require('node:events')
+
+    ;(async () => {
+      const server = http.createServer((req, res) => res.end('ok'))
+      server.listen(0)
+      await once(server, 'listening')
+
+      setGlobalDispatcher(new Agent())
+      const url = 'http://127.0.0.1:' + server.address().port
+      const res = await fetch(url)
+      process.stdout.write(await res.text())
+
+      server.close()
+    })().catch((err) => {
+      console.error(err?.cause?.stack || err?.stack || err)
+      process.exit(1)
+    })
+  `
+
+  const result = runNode(script)
+  assert.strictEqual(result.status, 0, result.stderr)
+  assert.strictEqual(result.stdout, 'ok')
+})
+
+test('setGlobalDispatcher mirrors a v1-compatible dispatcher on Node.js 22', () => {
+  const script = `
+    const { Agent, Dispatcher1Wrapper, setGlobalDispatcher } = require('./index.js')
+    const nodeMajor = Number(process.versions.node.split('.', 1)[0])
+
+    setGlobalDispatcher(new Agent())
+
+    if (nodeMajor !== 22) {
+      process.stdout.write('skipped')
+    } else {
+      const dispatcherV1 = globalThis[Symbol.for('undici.globalDispatcher.1')]
+
+      if (!(dispatcherV1 instanceof Dispatcher1Wrapper)) {
+        throw new Error('expected v1 global dispatcher to be a Dispatcher1Wrapper on Node.js 22')
+      }
+
+      process.stdout.write('mirrored')
+    }
+  `
+
+  const result = runNode(script)
+  assert.strictEqual(result.status, 0, result.stderr)
+
+  const expected = Number(process.versions.node.split('.', 1)[0]) === 22 ? 'mirrored' : 'skipped'
+  assert.strictEqual(result.stdout, expected)
+})
+
+test('Dispatcher1Wrapper bridges legacy handlers to a new Agent', () => {
+  const script = `
+    const { Agent, Dispatcher1Wrapper } = require('./index.js')
+    const http = require('node:http')
+    const { once } = require('node:events')
+
+    ;(async () => {
+      const server = http.createServer((req, res) => res.end('ok'))
+      server.listen(0)
+      await once(server, 'listening')
+
+      const dispatcherV1 = Symbol.for('undici.globalDispatcher.1')
+      globalThis[dispatcherV1] = new Dispatcher1Wrapper(new Agent())
+
+      const url = 'http://127.0.0.1:' + server.address().port
+      const res = await fetch(url)
+      process.stdout.write(await res.text())
+
+      server.close()
+    })().catch((err) => {
+      console.error(err?.cause?.stack || err?.stack || err)
+      process.exit(1)
+    })
+  `
+
+  const result = runNode(script)
+  assert.strictEqual(result.status, 0, result.stderr)
+  assert.strictEqual(result.stdout, 'ok')
+})

types/dispatcher1-wrapper.d.ts

@@ -0,0 +1,7 @@
+import Dispatcher from './dispatcher'
+
+export default Dispatcher1Wrapper
+
+declare class Dispatcher1Wrapper extends Dispatcher {
+  constructor (dispatcher: Dispatcher)
+}

types/index.d.ts

@@ -11,6 +11,7 @@ import H2CClient from './h2c-client'
 import buildConnector from './connector'
 import errors from './errors'
 import Agent from './agent'
+import Dispatcher1Wrapper from './dispatcher1-wrapper'
 import MockClient from './mock-client'
 import MockPool from './mock-pool'
 import MockAgent from './mock-agent'
@@ -43,7 +44,7 @@ export { Interceptable } from './mock-interceptor'
 
 declare function globalThisInstall (): void
 
-export { Dispatcher, BalancedPool, RoundRobinPool, Pool, Client, buildConnector, errors, Agent, request, stream, pipeline, connect, upgrade, setGlobalDispatcher, getGlobalDispatcher, setGlobalOrigin, getGlobalOrigin, interceptors, cacheStores, MockClient, MockPool, MockAgent, SnapshotAgent, MockCallHistory, MockCallHistoryLog, mockErrors, ProxyAgent, EnvHttpProxyAgent, RedirectHandler, DecoratorHandler, RetryHandler, RetryAgent, H2CClient, globalThisInstall as install }
+export { Dispatcher, BalancedPool, RoundRobinPool, Pool, Client, buildConnector, errors, Agent, Dispatcher1Wrapper, request, stream, pipeline, connect, upgrade, setGlobalDispatcher, getGlobalDispatcher, setGlobalOrigin, getGlobalOrigin, interceptors, cacheStores, MockClient, MockPool, MockAgent, SnapshotAgent, MockCallHistory, MockCallHistoryLog, mockErrors, ProxyAgent, EnvHttpProxyAgent, RedirectHandler, DecoratorHandler, RetryHandler, RetryAgent, H2CClient, globalThisInstall as install }
 export default Undici
 
 declare namespace Undici {
@@ -59,6 +60,7 @@ declare namespace Undici {
   const buildConnector: typeof import('./connector').default
   const errors: typeof import('./errors').default
   const Agent: typeof import('./agent').default
+  const Dispatcher1Wrapper: typeof import('./dispatcher1-wrapper').default
   const setGlobalDispatcher: typeof import('./global-dispatcher').setGlobalDispatcher
   const getGlobalDispatcher: typeof import('./global-dispatcher').getGlobalDispatcher
   const request: typeof import('./api').request