[world] Add WORKFLOW_NODE_HTTP to run the HTTP Worlds on node:http (#3461)

Author: VaguelySerious

.changeset/node-http-transport-flag.md

@@ -0,0 +1,7 @@
+---
+'@workflow/world-vercel': minor
+'@workflow/world-local': minor
+'@workflow/world': minor
+---
+
+Add `WORKFLOW_NODE_HTTP` to run the Vercel and Local World HTTP layers on Node's built-in `node:http` and `node:https` modules instead of their usual HTTP client library.

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -214,6 +214,24 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - Default: enabled
 - On the Vercel World, lets concurrent event-log requests share one HTTP/2 connection instead of one connection per in-flight request.
 - Set `0` to take the event-log requests off HTTP/2 entirely, back to one request per HTTP/1.1 connection. Use this as the kill switch if HTTP/2 turns out to be at fault for event delivery problems.
+- Has no effect when `WORKFLOW_NODE_HTTP` is enabled, which takes the whole HTTP/2 path away.
+
+### `WORKFLOW_NODE_HTTP`
+
+- Default: disabled
+- Makes the Vercel and Local Worlds issue their HTTP requests through Node's built-in `node:http` and `node:https` modules, instead of the HTTP client library those Worlds normally use.
+- Set `1` to switch to Node's modules. Read the trade-offs below first: they cost throughput on every deployment, which is why this is opt-in.
+- Use it when that library is not an option: a bundler that mangles it, a runtime that does not ship a working copy of it, or a transport-level fault you want to rule out. It is not reached by way of `fetch()` either, so a runtime whose `fetch()` is built on the same library is still covered.
+
+Node's own modules do less than the client they replace, so enabling this drops the per-call-site tuning the Worlds configure:
+
+- Event-log requests lose HTTP/2, so concurrent reads and writes no longer share one connection, and the enlarged HTTP/2 receive windows no longer apply. This is the largest difference, and it slows down replays that read a big event log. It does not apply to event writes on [`WORKFLOW_EVENTS_TRANSPORT=ws`](/docs/configuration/worlds#workflow_events_transport), which take neither transport.
+- Requests lose their transport-level retry. Failures still surface to the layers above, which retry event writes and redeliver queue messages, so nothing is silently dropped, but a failure that a same-connection retry would have hidden now costs a full redelivery.
+- Stream close loses its retry of retriable server errors. A transient failure at close can leave a stream marked closing until the run expires, where it would previously have resolved on the retry.
+
+Connection pooling, keep-alive, and the request, header, and body deadlines are preserved: pooling and keep-alive are configured on Node's agents, and the deadlines are passed per request — from the Local World's two queue timeouts, and on the Vercel World from the same defaults its HTTP client applies today. Queue sends are the exception in the other direction: that client takes no transport override, so its requests keep using the library either way.
+
+A `dispatcher` passed to `createVercelWorld()` still wins over this variable. The variable chooses which transport the World builds when you have not supplied one.
 
 ## Queue namespace
 

docs/content/worlds/v5/local.mdx

@@ -70,6 +70,10 @@ Maximum time in milliseconds to wait for a local queue handler to begin respondi
 
 Maximum gap in milliseconds between response body chunks from a local queue handler before redelivering the durable message. Default: `30000`. Set to `0` to disable.
 
+### `WORKFLOW_NODE_HTTP`
+
+Whether queue deliveries go out through Node's built-in `node:http` and `node:https` modules instead of the HTTP client library this World normally uses. Default: disabled. Set to `1` to switch to Node's modules. Socket pooling, keep-alive, and both timeouts above apply either way. See [`WORKFLOW_NODE_HTTP`](/docs/configuration/runtime-tuning#workflow_node_http) for what else changes.
+
 ### `WORKFLOW_LOCAL_RECOVER_ACTIVE_RUNS`
 
 Whether pending and running runs found in the data directory are re-enqueued when the world starts. Set to `0` or `false` to skip recovery and leave stale runs untouched. Default: `true`

packages/world-local/src/queue.test.ts

@@ -2,7 +2,7 @@ import { createServer, type Server } from 'node:http';
 import type { AddressInfo } from 'node:net';
 import { setWorkflowBasePath } from '@workflow/utils';
 import type { WorkflowInvokePayload } from '@workflow/world';
-import { MessageId, ValidQueueName } from '@workflow/world';
+import { MessageId, NODE_HTTP_ENV_VAR, ValidQueueName } from '@workflow/world';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { z } from 'zod/v4';
 import {
@@ -23,6 +23,19 @@ const workflowPayload: WorkflowInvokePayload = {
   stepName: 'test-step',
 };
 
+// The suite below covers the queue on undici, including the tests that stub
+// the global `fetch`. `WORKFLOW_NODE_HTTP` sends deliveries over `node:http`
+// instead, where stubbing `fetch` proves nothing, so pin the flag off rather
+// than tracking whichever way its default points. That mode has its own
+// describe at the end.
+beforeEach(() => {
+  vi.stubEnv(NODE_HTTP_ENV_VAR, '0');
+});
+
+afterEach(() => {
+  vi.unstubAllEnvs();
+});
+
 describe('zod v3/v4 schema compatibility (regression #1587)', () => {
   it('ValidQueueName and MessageId from @workflow/world parse correctly in z.object()', () => {
     const HeaderParser = z.object({
@@ -615,3 +628,95 @@ describe('queue transport timeouts', () => {
     }
   });
 });
+
+describe('node:http mode', () => {
+  let server: Server | undefined;
+
+  beforeEach(() => {
+    vi.stubEnv(NODE_HTTP_ENV_VAR, '1');
+  });
+
+  afterEach(async () => {
+    if (server !== undefined) {
+      const toClose = server;
+      server = undefined;
+      toClose.closeAllConnections();
+      await new Promise((resolve) => toClose.close(resolve));
+    }
+    vi.restoreAllMocks();
+    vi.unstubAllGlobals();
+  });
+
+  // The equivalence claim the flag makes: a delivery still goes out, still
+  // carries the VQS headers the handler reads, and still lands on the handler.
+  // The global `fetch` is stubbed to throw to prove the request left undici
+  // entirely rather than falling back to the runtime's own pool.
+  it('delivers without going through fetch', async () => {
+    const attempts: (string | undefined)[] = [];
+    server = createServer((request, response) => {
+      attempts.push(request.headers['x-vqs-message-attempt'] as string);
+      response.setHeader('content-type', 'application/json');
+      response.end(JSON.stringify({ ok: true }));
+    });
+    await new Promise<void>((resolve) => {
+      server?.listen(0, '127.0.0.1', resolve);
+    });
+    const { port } = server.address() as AddressInfo;
+
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(() => {
+        throw new Error('fetch must not be used under WORKFLOW_NODE_HTTP');
+      })
+    );
+
+    const localQueue = createQueue({ baseUrl: `http://127.0.0.1:${port}` });
+    try {
+      await localQueue.queue('__wkf_workflow_test' as any, workflowPayload);
+      await vi.waitFor(() => expect(attempts.length).toBe(1), {
+        timeout: 5_000,
+      });
+      expect(attempts[0]).toBe('1');
+    } finally {
+      await localQueue.close();
+    }
+  });
+
+  // Redelivery on a transport throw is the queue's own logic, not undici's, so
+  // it survives the switch. Node's client raises ECONNRESET where undici would
+  // have raised a TypeError wrapping UND_ERR_SOCKET; the delivery loop keys on
+  // neither, so both retry the same durable message.
+  it('still retries a transport throw', async () => {
+    let calls = 0;
+    server = createServer((request, response) => {
+      calls++;
+      if (calls < 3) {
+        request.socket.destroy();
+        return;
+      }
+      response.setHeader('content-type', 'application/json');
+      response.end(JSON.stringify({ ok: true }));
+    });
+    await new Promise<void>((resolve) => {
+      server?.listen(0, '127.0.0.1', resolve);
+    });
+    const { port } = server.address() as AddressInfo;
+
+    const localQueue = createQueue({ baseUrl: `http://127.0.0.1:${port}` });
+    try {
+      await localQueue.queue('__wkf_workflow_test' as any, workflowPayload);
+      await vi.waitFor(() => expect(calls).toBe(3), { timeout: 20_000 });
+    } finally {
+      await localQueue.close();
+    }
+  }, 30_000);
+
+  // close() owns a socket pool here too, just Node's rather than undici's. It
+  // still has to settle, still has to be idempotent, and still has to stop
+  // in-flight deliveries via the abort controller it owns.
+  it('closes idempotently', async () => {
+    const localQueue = createQueue({ baseUrl: 'http://localhost:3000' });
+    await expect(localQueue.close()).resolves.toBeUndefined();
+    await expect(localQueue.close()).resolves.toBeUndefined();
+  });
+});

packages/world-local/src/queue.ts

@@ -2,12 +2,18 @@ import { setTimeout } from 'node:timers/promises';
 import type { Transport } from '@vercel/queue';
 import { createWorkflowUrl } from '@workflow/utils';
 import {
+  isNodeHttpEnabled,
   MessageId,
   parseQueueName,
   type Queue,
   type QueuePrefix,
   ValidQueueName,
 } from '@workflow/world';
+import {
+  createNodeHttpAgents,
+  destroyNodeHttpAgents,
+  nodeHttpFetch,
+} from '@workflow/world/node-http.js';
 import { Sema } from 'async-sema';
 import { monotonicFactory } from 'ulid';
 import { Agent } from 'undici';
@@ -78,6 +84,11 @@ function envTimeoutMs(name: string, fallback: number): number {
  * Bounds a stalled delivery below the queue handler's retry horizon. A
  * transport timeout is retried by the delivery loop with the same durable
  * message; `0` remains available for applications that need unbounded calls.
+ *
+ * Both transports honour every field. Over undici this is the `Agent`'s own
+ * configuration; over `node:http` (`WORKFLOW_NODE_HTTP`) the two timeouts are
+ * passed per request and `connections` / `keepAliveTimeout` size the socket
+ * pool, so the two env vars tune a delivery identically either way.
  */
 export function getQueueAgentOptions() {
   return {
@@ -128,7 +139,17 @@ function isDetachedArrayBufferQueueError(error: unknown): boolean {
 }
 
 export function createQueue(config: Partial<Config>): LocalQueue {
-  const httpAgent = new Agent(getQueueAgentOptions());
+  // Exactly one of these is built, and close() shuts down whichever it is.
+  // Resolved once per queue rather than per delivery so a single queue never
+  // mixes transports mid-flight.
+  const agentOptions = getQueueAgentOptions();
+  const nodeHttpAgents = isNodeHttpEnabled()
+    ? createNodeHttpAgents({
+        maxSockets: agentOptions.connections,
+        keepAliveMs: agentOptions.keepAliveTimeout,
+      })
+    : undefined;
+  const httpAgent = nodeHttpAgents ? undefined : new Agent(agentOptions);
   const transport = new TypedJsonTransport();
   const generateId = monotonicFactory();
   const semaphore = new Sema(WORKFLOW_LOCAL_QUEUE_CONCURRENCY);
@@ -231,17 +252,24 @@ export function createQueue(config: Partial<Config>): LocalQueue {
               response = await directHandler(req);
             } else {
               const baseUrl = await resolveBaseUrl(config);
-              // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici v7 dispatcher types don't match @types/node's RequestInit
-              response = await fetch(
-                createWorkflowUrl(baseUrl, { type: 'flow' }),
-                {
-                  method: 'POST',
-                  duplex: 'half',
-                  dispatcher: httpAgent,
-                  headers,
-                  body,
-                } as any
-              );
+              const url = createWorkflowUrl(baseUrl, { type: 'flow' });
+              response = nodeHttpAgents
+                ? await nodeHttpFetch(url, {
+                    method: 'POST',
+                    headers: new Headers(headers),
+                    body,
+                    agents: nodeHttpAgents,
+                    headersTimeoutMs: agentOptions.headersTimeout,
+                    bodyTimeoutMs: agentOptions.bodyTimeout,
+                  })
+                : // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici v7 dispatcher types don't match @types/node's RequestInit
+                  await fetch(url, {
+                    method: 'POST',
+                    duplex: 'half',
+                    dispatcher: httpAgent,
+                    headers,
+                    body,
+                  } as any);
             }
             delivery++;
             text = await response.text();
@@ -433,7 +461,8 @@ export function createQueue(config: Partial<Config>): LocalQueue {
       // may close the queue more than once.
       if (closeSignal.aborted) return;
       closeController.abort();
-      await httpAgent.close();
+      if (nodeHttpAgents) destroyNodeHttpAgents(nodeHttpAgents);
+      await httpAgent?.close();
     },
   };
 }

packages/world-vercel/src/events-v4.test.ts

@@ -8,9 +8,10 @@ import {
   WorkflowWorldError,
 } from '@workflow/errors';
 import type { AnyEventRequest } from '@workflow/world';
+import { NODE_HTTP_ENV_VAR } from '@workflow/world';
 import { decode, encode } from 'cbor-x';
 import { MockAgent } from 'undici';
-import { afterEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { splitEventDataForV4 } from './events.js';
 import {
   createWorkflowRunEventV4,
@@ -1506,7 +1507,15 @@ describe('v4 POST frame meta forwards every field the splitter produces', () =>
  * `fetchV4` the recycler is never told anything and the pool lives forever.
  */
 describe('v4 transport reports failures to the events recycler', () => {
+  // There is only an undici pool to retire while the adapter owns one:
+  // `WORKFLOW_NODE_HTTP` takes the request off undici and makes
+  // getEventsDispatcher return `undefined`, so pin the flag off here.
+  beforeEach(() => {
+    vi.stubEnv(NODE_HTTP_ENV_VAR, '0');
+  });
+
   afterEach(() => {
+    vi.unstubAllEnvs();
     vi.restoreAllMocks();
   });