From b1946788557888803c9eaab9912238d8e476beb1 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Wed, 29 Jul 2026 14:38:17 -0700 Subject: [PATCH] [world-vercel] Make HTTP/2 actually multiplex on the events path (#3190) (cherry picked from commit 34975f6b7dd8e0dad874e852eac04c9652f5971d) Signed-off-by: Pranay Prakash --- .changeset/h2-events-multiplexing.md | 5 + .../docs/v5/configuration/runtime-tuning.mdx | 8 + packages/world-vercel/src/http-client.test.ts | 223 +++++++++++++++ packages/world-vercel/src/http-client.ts | 254 ++++++++++++++++-- 4 files changed, 468 insertions(+), 22 deletions(-) create mode 100644 .changeset/h2-events-multiplexing.md diff --git a/.changeset/h2-events-multiplexing.md b/.changeset/h2-events-multiplexing.md new file mode 100644 index 0000000000..e516cf4334 --- /dev/null +++ b/.changeset/h2-events-multiplexing.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-vercel': patch +--- + +Event-log requests now multiplex over a single HTTP/2 connection instead of opening one connection per in-flight request diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 932a4ac38c..2f3e68e8c3 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -115,6 +115,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Debug log filter with wildcards and negation. - Examples: `workflow:*`, `workflow:*,-workflow:telemetry:*`. +## Transport + +### `WORKFLOW_H2_MULTIPLEX` + +- 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 send one event request per connection. + ## Queue namespace ### `WORKFLOW_QUEUE_NAMESPACE` diff --git a/packages/world-vercel/src/http-client.test.ts b/packages/world-vercel/src/http-client.test.ts index 29db095dbf..e6aa197ece 100644 --- a/packages/world-vercel/src/http-client.test.ts +++ b/packages/world-vercel/src/http-client.test.ts @@ -4,12 +4,15 @@ import type { TLSSocket } from 'node:tls'; import { Agent } from 'undici'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { + createEventsDispatcher, + createStreamDispatcher, DEFAULT_AGENT_OPTIONS, EVENTS_AGENT_OPTIONS, getDispatcher, getEventsDispatcher, getStreamCloseDispatcher, getStreamDispatcher, + STREAM_AGENT_OPTIONS, STREAM_CLOSE_RETRY_OPTIONS, STREAM_RETRY_OPTIONS, } from './http-client.js'; @@ -94,8 +97,25 @@ describe('agent transport', () => { // Flipping either silently would regress one side or the other. it('enables HTTP/2 for the events API only', () => { expect(EVENTS_AGENT_OPTIONS.allowH2).toBe(true); + expect(STREAM_AGENT_OPTIONS.allowH2).toBe(true); expect(DEFAULT_AGENT_OPTIONS.allowH2).toBe(false); }); + + // `allowH2` alone buys nothing: undici gates in-flight requests per + // connection on `pipelining`, so `pipelining: 1` reduces an H2 agent to H1 + // behavior (one stream per connection). These two constants are the + // difference between multiplexing and not — see EVENTS_AGENT_OPTIONS. + it('gives the events agent a pipelining depth that permits multiplexing', () => { + expect(EVENTS_AGENT_OPTIONS.pipelining).toBeGreaterThan(1); + }); + + // Inverse guard: stream appends are not idempotent, so they must NOT + // multiplex — one connection-level failure would fail (and retry) several + // appends at once. See STREAM_AGENT_OPTIONS. + it('keeps stream writes and the H1 default at one request per connection', () => { + expect(STREAM_AGENT_OPTIONS.pipelining).toBe(1); + expect(DEFAULT_AGENT_OPTIONS.pipelining).toBe(1); + }); }); // Self-signed cert for localhost, valid 100 years. Generated with: @@ -199,3 +219,206 @@ describe('HTTP/2 over global fetch with an undici dispatcher', () => { expect(negotiatedAlpn).toBe('h2'); }); }); + +// Negotiating h2 is not the same as using it. This measures the property the +// events agent actually exists for: concurrent POSTs sharing ONE connection as +// parallel H2 streams. It is the regression test the config-only assertions +// above cannot be — before the pipelining + interceptor fix, `allowH2` was true +// and ALPN was h2, yet 16 concurrent requests still produced 8 serialized +// requests over 8 TCP connections, exactly like the H1 agent. +describe('HTTP/2 multiplexing (events vs stream-write agents)', () => { + const CONCURRENCY = 16; + + let server: Http2SecureServer; + let port: number; + let sessions: number; + let maxConcurrentStreams: number; + let inFlight: number; + let receivedBodies: string[]; + let release: Array<() => void>; + let flakyAttempts: number; + + /** + * Holds every request open until `CONCURRENCY` of them are in flight, so peak + * concurrency is observed rather than timed. A periodic flush (see `burst`) + * drains whatever is waiting when that target is never reached — which is the + * expected outcome for a non-multiplexing agent, and must fail the assertion + * rather than hang the test. + */ + function onArrival(path: string): Promise { + // The pool-warming request is not part of the barrier — it must complete on + // its own so the burst starts from an established session. + if (!path.startsWith('/req-')) return Promise.resolve(); + return new Promise((resolve) => { + inFlight++; + maxConcurrentStreams = Math.max(maxConcurrentStreams, inFlight); + release.push(resolve); + if (release.length >= CONCURRENCY) { + for (const r of release.splice(0)) r(); + } + }); + } + + beforeAll(async () => { + server = createSecureServer({ key: TEST_KEY, cert: TEST_CERT }); + server.on('session', (session) => { + sessions++; + // Agents are closed while the pool still holds idle sessions; the + // resulting resets are expected teardown noise, not test failures. + session.on('error', () => undefined); + }); + server.on('sessionError', () => undefined); + server.on('clientError', () => undefined); + server.on('stream', (stream, headers) => { + stream.on('error', () => undefined); + const chunks: Buffer[] = []; + stream.on('data', (c: Buffer) => chunks.push(c)); + stream.on('end', () => { + void (async () => { + const path = String(headers[':path']); + receivedBodies.push(Buffer.concat(chunks).toString()); + await onArrival(path); + if (path.startsWith('/req-')) inFlight--; + // `/flaky` fails once so RetryAgent re-dispatches it. + if (path === '/flaky' && ++flakyAttempts === 1) { + stream.respond({ ':status': 503 }); + stream.end('retry me'); + return; + } + stream.respond({ ':status': 200 }); + stream.end(path); + })(); + }); + }); + await new Promise((resolve) => { + server.listen(0, '127.0.0.1', resolve); + }); + port = (server.address() as AddressInfo).port; + }); + + afterAll(async () => { + await new Promise((resolve) => { + server.close(() => resolve()); + }); + }); + + /** Loopback TLS escape hatch — the only deviation from production wiring. */ + const LOOPBACK = { connect: { rejectUnauthorized: false } }; + + async function burst(dispatcher: unknown) { + sessions = 0; + maxConcurrentStreams = 0; + inFlight = 0; + receivedBodies = []; + release = []; + flakyAttempts = 0; + // Warm the pool so connection setup isn't conflated with the stream gate: + // a cold burst races ALPN negotiation and fans out across connections. + await fetch(`https://127.0.0.1:${port}/warm`, { + dispatcher, + method: 'POST', + body: 'warm', + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici dispatcher type doesn't match @types/node's RequestInit + } as any); + const sessionsAfterWarm = sessions; + // Repeating, not one-shot: an agent that caps in-flight requests below + // CONCURRENCY delivers the burst in several waves, and every wave needs + // draining or the remainder blocks forever. + const timer = setInterval(() => { + for (const r of release.splice(0)) r(); + }, 250); + const bodies = await Promise.all( + Array.from({ length: CONCURRENCY }, (_, i) => + fetch(`https://127.0.0.1:${port}/req-${i}`, { + method: 'POST', + body: JSON.stringify({ i }), + dispatcher, + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici dispatcher type doesn't match @types/node's RequestInit + } as any).then((r) => r.text()) + ) + ); + clearInterval(timer); + return { bodies, sessionsAfterWarm }; + } + + it('multiplexes concurrent event writes onto a single connection', async () => { + // The real production factory — so dropping the interceptor from + // createEventsDispatcher fails here, not just changing the constants. + const agent = createEventsDispatcher(LOOPBACK); + try { + const { bodies, sessionsAfterWarm } = await burst(agent); + + expect(maxConcurrentStreams).toBe(CONCURRENCY); + // No new TCP/TLS session beyond the warmed one — the whole point. + expect(sessions).toBe(sessionsAfterWarm); + // Re-buffering the body must not corrupt or cross-wire payloads. + expect(bodies.sort()).toEqual( + Array.from({ length: CONCURRENCY }, (_, i) => `/req-${i}`).sort() + ); + expect(receivedBodies.filter((b) => b !== 'warm').sort()).toEqual( + Array.from({ length: CONCURRENCY }, (_, i) => + JSON.stringify({ i }) + ).sort() + ); + } finally { + await agent.close(); + } + }); + + it('does not multiplex stream writes (non-idempotent appends stay isolated)', async () => { + const agent = createStreamDispatcher(STREAM_RETRY_OPTIONS, LOOPBACK); + try { + await burst(agent); + // Bounded by the pool size, not by CONCURRENCY: each connection carries + // at most one append, so a reset can only ever fail one write. + expect(maxConcurrentStreams).toBeLessThanOrEqual( + STREAM_AGENT_OPTIONS.connections + ); + expect(maxConcurrentStreams).toBeLessThan(CONCURRENCY); + } finally { + await agent.close(); + } + }); + + it('resends the full body when a re-buffered request is retried', async () => { + // The interceptor consumes the request body to make it multiplexable, but + // RetryAgent re-dispatches with the *original* (now exhausted) stream. If the + // drained buffer were not reused, the retry would arrive with an empty body. + const agent = createEventsDispatcher(LOOPBACK); + receivedBodies = []; + flakyAttempts = 0; + const payload = JSON.stringify({ chunk: 'x'.repeat(64) }); + try { + const response = await fetch(`https://127.0.0.1:${port}/flaky`, { + method: 'PUT', + body: payload, + dispatcher: agent, + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici dispatcher type doesn't match @types/node's RequestInit + } as any); + expect(response.status).toBe(200); + expect(flakyAttempts).toBe(2); + expect(receivedBodies).toEqual([payload, payload]); + } finally { + await agent.close(); + } + }); + + it('WORKFLOW_H2_MULTIPLEX=0 falls back to one request per connection', async () => { + const previous = process.env.WORKFLOW_H2_MULTIPLEX; + process.env.WORKFLOW_H2_MULTIPLEX = '0'; + // Read when the dispatcher is built, so the kill switch only takes effect + // for agents created after it is set. + const agent = createEventsDispatcher(LOOPBACK); + try { + await burst(agent); + expect(maxConcurrentStreams).toBeLessThan(CONCURRENCY); + } finally { + await agent.close(); + if (previous === undefined) { + delete process.env.WORKFLOW_H2_MULTIPLEX; + } else { + process.env.WORKFLOW_H2_MULTIPLEX = previous; + } + } + }); +}); diff --git a/packages/world-vercel/src/http-client.ts b/packages/world-vercel/src/http-client.ts index 2cb181d786..431fd52e47 100644 --- a/packages/world-vercel/src/http-client.ts +++ b/packages/world-vercel/src/http-client.ts @@ -1,4 +1,4 @@ -import { Agent, RetryAgent, type RetryHandler } from 'undici'; +import { Agent, type Dispatcher, RetryAgent, type RetryHandler } from 'undici'; import type { APIConfig } from './utils.js'; let _dispatcher: RetryAgent | undefined; @@ -6,15 +6,25 @@ let _eventsDispatcher: RetryAgent | undefined; let _streamDispatcher: RetryAgent | undefined; let _streamCloseDispatcher: RetryAgent | undefined; -/** Shared between both agents — connection pooling and H1 pipelining tuning. */ +/** + * Shared between all agents — connection pooling only. `pipelining` is + * deliberately NOT set here: undici overloads that single option to mean both + * "H1 pipelining depth" and "max in-flight H2 streams per connection", and the + * two paths want opposite values. Each agent sets it explicitly below. + */ const BASE_AGENT_OPTIONS = { connections: 8, keepAliveTimeout: 10_000, - // HTTP/1.1 pipelining is disabled (pipelining: 1) because it causes - // head-of-line blocking that deadlocks the webhook respondWith mechanism. - pipelining: 1, }; +/** + * In-flight H2 streams allowed per connection. Matches undici's + * `maxConcurrentStreams` default (100), which is the real ceiling once the + * server's SETTINGS_MAX_CONCURRENT_STREAMS is known — so this only has to be + * large enough not to be the binding constraint. + */ +const H2_MAX_IN_FLIGHT_STREAMS = 100; + /** * Options for the default undici Agent — the queue client (webhook * respondWith), v3 `makeRequest`, deployment resolution, and run-key fetch. @@ -28,11 +38,14 @@ const BASE_AGENT_OPTIONS = { export const DEFAULT_AGENT_OPTIONS = { ...BASE_AGENT_OPTIONS, allowH2: false, + // HTTP/1.1 pipelining is disabled (pipelining: 1) because it causes + // head-of-line blocking that deadlocks the webhook respondWith mechanism. + pipelining: 1, } as const; /** * Options for the events API undici Agent. Exported so tests can assert that - * HTTP/2 stays enabled. + * HTTP/2 stays enabled *and* that it is actually configured to multiplex. * * The v4 events endpoints are the hottest path (an event write per step * transition, plus event-log reads on replay) and are plain request/response — @@ -41,10 +54,47 @@ export const DEFAULT_AGENT_OPTIONS = { * removes per-request connection setup and head-of-line blocking here. * Re-enabling H2 more broadly is gated on resolving those issues (notably the * earlier SvelteKit-on-Vercel-prod hang). + * + * `pipelining` must be set for H2 to multiplex at all. undici gates in-flight + * requests per connection on `getPipelining(client)`, which reads + * `client[kPipelining] ?? httpContext.defaultPipelining ?? 1`. `client-h2.js` + * sets `defaultPipelining: Infinity`, but the Client constructor coerces + * `pipelining` to a number (`pipelining != null ? pipelining : 1`), so + * `kPipelining` is never nullish and H2's Infinity is unreachable — leaving one + * in-flight stream per connection. Before this was set, the H2 agent behaved + * byte-for-byte like the H1 agent: 16 concurrent requests produced 8 in-flight + * requests over 8 TCP connections. See nodejs/undici#4143. */ export const EVENTS_AGENT_OPTIONS = { ...BASE_AGENT_OPTIONS, allowH2: true, + pipelining: H2_MAX_IN_FLIGHT_STREAMS, +} as const; + +/** + * Options for the stream write/close Agents. H2 is enabled (these send a + * fully-buffered body, or none, so they avoid the duplex-streaming issues that + * keep the long-lived live-read on plain `fetch`), but multiplexing is + * deliberately left OFF — `pipelining: 1`, one in-flight request per + * connection. + * + * Stream appends are not idempotent. Multiplexing N appends onto one connection + * makes a single RST_STREAM / GOAWAY / socket reset fail all N at once, and + * STREAM_RETRY_OPTIONS retries PUT on exactly those transient `errorCodes` — so + * a connection-level blip would resend chunks the server may already have + * applied and duplicate them. Serializing keeps the existing + * one-request-per-connection failure isolation that policy was written against. + * + * Note this is currently belt-and-braces: undici's H2 `busy()` check already + * serializes non-idempotent requests (`client-h2.js`: `if + * (request.idempotent === false) return true`), so PUTs would not multiplex even + * at a higher pipelining value. Setting it explicitly means the safety property + * does not silently depend on that upstream detail. + */ +export const STREAM_AGENT_OPTIONS = { + ...BASE_AGENT_OPTIONS, + allowH2: true, + pipelining: 1, } as const; const RETRY_AGENT_OPTIONS: RetryHandler.RetryOptions = { @@ -108,6 +158,103 @@ export const STREAM_CLOSE_RETRY_OPTIONS: RetryHandler.RetryOptions = { statusCodes: [429, 500, 502, 503, 504], }; +/** + * Largest body the H2 multiplexing interceptor will re-buffer. Event frames are + * small CBOR payloads; the cap is a guard against an unexpectedly large write + * being held in memory twice, not a tuning knob. + */ +const H2_REBUFFER_MAX_BYTES = 1 << 20; // 1 MiB + +/** Set `WORKFLOW_H2_MULTIPLEX=0` to fall back to one request per connection. */ +function h2MultiplexEnabled(): boolean { + return process.env.WORKFLOW_H2_MULTIPLEX !== '0'; +} + +function contentLength(headers: unknown): number { + if (!headers || typeof headers !== 'object' || Array.isArray(headers)) { + return Number.NaN; + } + const record = headers as Record; + const raw = record['content-length'] ?? record['Content-Length']; + return typeof raw === 'string' || typeof raw === 'number' + ? Number(raw) + : Number.NaN; +} + +/** + * Undici interceptor that lets the events API actually multiplex over H2. + * + * `pipelining` (see EVENTS_AGENT_OPTIONS) is necessary but not sufficient: + * undici's H2 `busy()` check reports the connection busy — serializing the + * request behind whatever is in flight — for two more reasons, both of which + * every events request trips. + * + * 1. Non-idempotent method. `client-h2.js` returns busy when + * `request.idempotent === false`, and undici only treats GET/HEAD as + * idempotent by default (`core/request.js`), so every event-write POST + * serializes. We mark them idempotent for *concurrency* purposes only: in + * undici 7 the flag feeds nothing but the H1/H2 `busy()` gates — it does not + * cause resends. Retries are governed solely by RetryAgent, whose + * `methods` default (`['GET','HEAD','OPTIONS','PUT','DELETE','TRACE']`) + * excludes POST, so an event write is still never replayed. See + * nodejs/undici#5390. + * + * 2. Streamed request body. `client-h2.js` also returns busy for a body that is + * a stream / async iterable, because such a body can error mid-flight and + * take unrelated in-flight requests down with it. events-v4 hands us a fully + * materialized `Uint8Array`, but it dispatches through the global `fetch` + * (deliberately — that is what keeps v4 traffic visible in Vercel's outgoing + * -requests view, see events-v4.ts), and `fetch` converts every body into an + * async iterable on the way down. Draining it back into a Buffer restores the + * buffered-body shape undici needs, at the cost of one copy of an + * already-in-memory payload. Bodies without a usable `content-length`, or + * above H2_REBUFFER_MAX_BYTES, are passed through untouched (and stay + * serialized) rather than buffered blind. + * + * Without both of these, `pipelining` alone leaves the events agent at one + * in-flight request per connection. + */ +export function h2MultiplexInterceptor( + dispatch: Dispatcher['dispatch'] +): Dispatcher['dispatch'] { + return (opts, handler) => { + const body = opts.body; + const isAsyncIterable = + !!body && + typeof body !== 'string' && + !Buffer.isBuffer(body) && + typeof (body as unknown as Record)[ + Symbol.asyncIterator + ] === 'function'; + const length = contentLength(opts.headers); + + if (!isAsyncIterable || !(length >= 0) || length > H2_REBUFFER_MAX_BYTES) { + return dispatch({ ...opts, idempotent: true }, handler); + } + + // Drain asynchronously, then dispatch. Returning `true` reports "no + // backpressure", which is accurate: the request is accepted, and the pool + // gate we are lifting is exactly the one that would have reported drain. + void (async () => { + try { + const chunks: Buffer[] = []; + for await (const chunk of body as AsyncIterable) { + chunks.push(Buffer.from(chunk)); + } + dispatch( + { ...opts, body: Buffer.concat(chunks), idempotent: true }, + handler + ); + } catch (error) { + // Surface a drain failure the way undici would have surfaced a body + // error, so the awaiting fetch() rejects instead of hanging. + handler.onError?.(error as Error); + } + })(); + return true; + }; +} + /** * Resolves the undici dispatcher for a request: the caller's override, or the * shared default agent (HTTP/1.1). @@ -147,12 +294,79 @@ export function getStreamCloseDispatcher(config?: APIConfig): unknown { /** Build a shared undici RetryAgent wrapping an Agent with the given options. */ function makeRetryDispatcher( - agentOptions: typeof DEFAULT_AGENT_OPTIONS | typeof EVENTS_AGENT_OPTIONS, + agentOptions: typeof DEFAULT_AGENT_OPTIONS, retryOptions: RetryHandler.RetryOptions ): RetryAgent { return new RetryAgent(new Agent(agentOptions), retryOptions); } +/** + * Builds the events-API dispatcher: the H2 agent plus the interceptor that makes + * H2 actually multiplex. Exported (rather than only reachable through the + * `getEventsDispatcher` singleton) so a test can exercise this exact wiring + * against a loopback server via `agentOverrides` — asserting on + * EVENTS_AGENT_OPTIONS alone cannot catch the composition being dropped. + */ +export function createEventsDispatcher( + agentOverrides?: Partial +): RetryAgent { + const agent = new RetryAgent( + new Agent({ ...EVENTS_AGENT_OPTIONS, ...agentOverrides }), + RETRY_AGENT_OPTIONS + ); + if (!h2MultiplexEnabled()) { + return agent; + } + // The interceptor wraps the RetryAgent (rather than the Agent inside it) so + // that retries re-send the *drained* body. RetryHandler captures its own copy + // of the request body up front — `wrapRequestBody` (undici core/util.js) hands + // an async-iterable body to a fresh `BodyAsyncIterable`. Composed inside, that + // copy would wrap the stream the interceptor is about to consume, so a retry + // would re-iterate an exhausted stream and send an empty body. Composed + // outside, RetryHandler captures the Buffer and replays it verbatim. + return withBoundLifecycle( + agent, + agent.compose(h2MultiplexInterceptor) as unknown as RetryAgent + ); +} + +/** + * Restores `close()`/`destroy()` on a composed dispatcher. + * + * `Dispatcher.compose()` returns `new Proxy(this, { get: (t, k) => k === + * 'dispatch' ? composed : t[k] })`, so every other method comes back unbound and + * runs with the Proxy as `this`. For a RetryAgent that means `close()` throws + * `TypeError: Cannot read private member #agent from an object whose class did + * not declare it`. Binding the lifecycle methods to the real instance keeps the + * composed dispatcher disposable. + */ +function withBoundLifecycle( + agent: RetryAgent, + composed: RetryAgent +): RetryAgent { + return new Proxy(composed, { + get: (target, key) => + key === 'close' || key === 'destroy' + ? (agent[key] as (...args: unknown[]) => unknown).bind(agent) + : target[key as keyof RetryAgent], + }); +} + +/** + * Builds a stream write/close dispatcher. Exported for the same reason as + * `createEventsDispatcher` — so a test can assert the inverse property, that + * these deliberately do NOT multiplex. + */ +export function createStreamDispatcher( + retryOptions: RetryHandler.RetryOptions, + agentOverrides?: Partial +): RetryAgent { + return new RetryAgent( + new Agent({ ...STREAM_AGENT_OPTIONS, ...agentOverrides }), + retryOptions + ); +} + /** * Returns the shared default RetryAgent. * @@ -172,13 +386,13 @@ function getDefaultDispatcher(): RetryAgent { /** * Returns the shared HTTP/2 RetryAgent used by the v4 events API. Same retry / - * pooling behavior as the default dispatcher, but with `allowH2` enabled. + * pooling behavior as the default dispatcher, but with `allowH2` enabled, a + * pipelining depth that lets H2 actually multiplex (EVENTS_AGENT_OPTIONS), and + * the interceptor that lifts undici's remaining two per-request H2 busy gates + * (h2MultiplexInterceptor). */ function getDefaultEventsDispatcher(): RetryAgent { - _eventsDispatcher ??= makeRetryDispatcher( - EVENTS_AGENT_OPTIONS, - RETRY_AGENT_OPTIONS - ); + _eventsDispatcher ??= createEventsDispatcher(); return _eventsDispatcher; } @@ -191,22 +405,18 @@ function getDefaultEventsDispatcher(): RetryAgent { * chunk was not persisted — and never on 5xx or other 4xx, where a retry could * duplicate an already-applied write. It opts into H2 (the write/close requests * send a fully-buffered body, or none, so they don't hit the duplex-streaming H2 - * issues that keep the long-lived live-read on plain `fetch`) by reusing the - * events agent's H2 / pooling options. + * issues that keep the long-lived live-read on plain `fetch`) via + * STREAM_AGENT_OPTIONS — which, unlike the events agent, keeps multiplexing off + * so one connection-level failure cannot fail (and thus retry) several appends + * at once. */ function getDefaultStreamDispatcher(): RetryAgent { - _streamDispatcher ??= makeRetryDispatcher( - EVENTS_AGENT_OPTIONS, - STREAM_RETRY_OPTIONS - ); + _streamDispatcher ??= createStreamDispatcher(STREAM_RETRY_OPTIONS); return _streamDispatcher; } /** Shared agent for the idempotent stream close (5xx retriable). */ function getDefaultStreamCloseDispatcher(): RetryAgent { - _streamCloseDispatcher ??= makeRetryDispatcher( - EVENTS_AGENT_OPTIONS, - STREAM_CLOSE_RETRY_OPTIONS - ); + _streamCloseDispatcher ??= createStreamDispatcher(STREAM_CLOSE_RETRY_OPTIONS); return _streamCloseDispatcher; }