Merge branch 'main' into jrvb/h2-transport-integration

# Conflicts:
#	loadtest/sse-test.ts
#	src/lib/h2-transport/index.ts
#	src/lib/h2-transport/pool.ts
#	src/lib/h2-transport/response.ts
#	src/lib/h2-transport/session.ts
#	tests/lib/h2-transport.test.ts

Author: Reflex

loadtest/sse-test.ts

@@ -119,6 +119,7 @@ async function main() {
   console.log('PASS');
 
   console.log('\n=== All SSE tests PASSED ===');
+  await h2Fetch.close();
   close();
 
   // Cleanup

src/lib/h2-transport/index.ts

@@ -36,20 +36,27 @@ function checkNodeVersion(): void {
 
 export type { H2PoolOptions as H2FetchOptions };
 
-function normalizeBody(body: unknown): string | Buffer | null {
+export type H2Fetch = Fetch & {
+  close: () => Promise<void>;
+};
+
+async function bufferReadable(stream: Readable): Promise<Buffer> {
+  const chunks: Buffer[] = [];
+  for await (const chunk of stream) {
+    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+  }
+  return Buffer.concat(chunks);
+}
+
+async function normalizeBody(body: unknown): Promise<string | Buffer | null> {
   if (body == null) return null;
   if (typeof body === 'string') return body;
   if (Buffer.isBuffer(body)) return body;
   if (body instanceof MultipartBody) return normalizeBody((body as MultipartBody).body);
   if (body instanceof Readable) {
-    // Multipart bodies arrive as Node Readable. For H2, we need to buffer them
-    // since session.request() takes string | Buffer. The Readable is a
-    // FormDataEncoder stream with a known content-length, so buffering is safe.
-    // Streaming uploads could be added later if needed.
-    throw new Error(
-      'h2-transport: streaming request bodies (Readable) are not yet supported. ' +
-        'Use a string or Buffer body.',
-    );
+    // Multipart bodies arrive as a FormDataEncoder Readable with a known
+    // content-length. session.request() takes string | Buffer, so buffer here.
+    return bufferReadable(body);
   }
   if (ArrayBuffer.isView(body)) return Buffer.from(body.buffer, body.byteOffset, body.byteLength);
   if (body instanceof ArrayBuffer) return Buffer.from(body);
@@ -62,7 +69,7 @@ function normalizeBody(body: unknown): string | Buffer | null {
  * Compatible with the SDK's `makeHttp2Fetch` interface: called with no arguments
  * for `http2: true`, or with options for `http2: <H2FetchOptions>`.
  */
-export function createH2Fetch(options?: H2PoolOptions): Fetch {
+export function createH2Fetch(options?: H2PoolOptions): H2Fetch {
   checkNodeVersion();
   const pools = new Map<string, H2Pool>();
 
@@ -75,7 +82,7 @@ export function createH2Fetch(options?: H2PoolOptions): Fetch {
     return pool;
   }
 
-  return async (url, init) => {
+  const h2Fetch = (async (url, init) => {
     const {
       agent: _ignored, // node-fetch artifact injected by core.ts
       body: rawBody,
@@ -86,7 +93,7 @@ export function createH2Fetch(options?: H2PoolOptions): Fetch {
 
     const parsed = typeof url === 'string' ? new URL(url) : new URL(url.toString());
     const path = parsed.pathname + parsed.search;
-    const body = normalizeBody(rawBody);
+    const body = await normalizeBody(rawBody);
 
     const reqHeaders: Record<string, string> = {};
     if (rawHeaders) {
@@ -103,5 +110,13 @@ export function createH2Fetch(options?: H2PoolOptions): Fetch {
 
     const pool = getPool(parsed.origin);
     return pool.request(path, method.toUpperCase(), reqHeaders, body, signal) as any;
+  }) as H2Fetch;
+
+  h2Fetch.close = async () => {
+    const closeTasks = [...pools.values()].map((pool) => pool.close());
+    pools.clear();
+    await Promise.all(closeTasks);
   };
+
+  return h2Fetch;
 }

src/lib/h2-transport/pool.ts

@@ -8,6 +8,7 @@ export interface H2PoolOptions extends H2SessionOptions {
 
 const DEFAULT_MIN_CONNECTIONS = 4;
 const DEFAULT_MAX_CONNECTIONS = 20;
+const RETRYABLE_METHODS = new Set(['GET', 'HEAD', 'OPTIONS', 'PUT', 'DELETE']);
 
 interface QueuedRequest {
   path: string;
@@ -46,9 +47,16 @@ export class H2Pool {
     if (this._initialized) return;
     if (this._initPromise) return this._initPromise;
 
-    this._initPromise = this._createInitialSessions();
-    await this._initPromise;
-    this._initialized = true;
+    const attempt = this._createInitialSessions();
+    this._initPromise = attempt;
+    try {
+      await attempt;
+      this._initialized = true;
+    } catch (err) {
+      // Don't cache the failure — let subsequent requests retry initialization.
+      if (this._initPromise === attempt) this._initPromise = null;
+      throw err;
+    }
   }
 
   private async _createInitialSessions(): Promise<void> {
@@ -134,7 +142,7 @@ export class H2Pool {
     try {
       return await session.request(path, method, headers, body, signal);
     } catch (err: any) {
-      if (session.state !== SessionState.READY) {
+      if (session.state !== SessionState.READY && RETRYABLE_METHODS.has(method)) {
         // Session went away (GOAWAY, etc). Re-enter the pool for a retry.
         return this._enqueueRequest(path, method, headers, body, signal);
       }
@@ -217,14 +225,14 @@ export class H2Pool {
 
   async close(): Promise<void> {
     this._closed = true;
-    for (const s of this._sessions) {
-      s.close();
-    }
+    const sessions = [...this._sessions];
     this._sessions.length = 0;
 
     for (const req of this._queue) {
       req.reject(new Error('Pool is closed'));
     }
     this._queue.length = 0;
+
+    await Promise.all(sessions.map((s) => s.close()));
   }
 }

src/lib/h2-transport/response.ts

@@ -1,3 +1,4 @@
+import { Blob } from 'node:buffer';
 import { H2Headers } from './headers';
 
 /**
@@ -53,4 +54,9 @@ export class H2Response {
     const buf = await this._consumeBody();
     return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength) as ArrayBuffer;
   }
+
+  async blob(): Promise<Blob> {
+    const buf = await this._consumeBody();
+    return new Blob([buf], { type: this.headers.get('content-type') ?? undefined });
+  }
 }

src/lib/h2-transport/session.ts

@@ -19,6 +19,8 @@ export interface H2SessionOptions {
 
 const DEFAULT_CONNECT_TIMEOUT = 30_000;
 
+const createAbortError = (): Error => Object.assign(new Error('Request aborted'), { name: 'AbortError' });
+
 export class H2Session {
   readonly origin: string;
 
@@ -65,14 +67,16 @@ export class H2Session {
 
       session.on('connect', () => {
         clearTimeout(timeout);
-        const serverMax = session.remoteSettings?.maxConcurrentStreams;
-        if (serverMax != null && !this._opts.maxConcurrentStreams) {
-          this._maxConcurrentStreams = serverMax;
-        }
         this._state = SessionState.READY;
         resolve();
       });
 
+      session.on('remoteSettings', (settings) => {
+        if (settings.maxConcurrentStreams != null && !this._opts.maxConcurrentStreams) {
+          this._maxConcurrentStreams = settings.maxConcurrentStreams;
+        }
+      });
+
       session.on('error', (err) => {
         clearTimeout(timeout);
         if (this._state === SessionState.CONNECTING) {
@@ -127,10 +131,11 @@ export class H2Session {
       this._activeStreams++;
 
       let settled = false;
+      let cleaned = false;
 
       const onAbort = () => {
-        if (!settled) {
-          stream.destroy(new Error('Request aborted'));
+        if (!cleaned) {
+          stream.destroy(createAbortError());
         }
       };
 
@@ -139,23 +144,27 @@ export class H2Session {
           stream.on('error', () => {});
           stream.destroy();
           this._activeStreams--;
-          reject(new Error('Request aborted'));
+          reject(createAbortError());
           return;
         }
         signal.addEventListener('abort', onAbort, { once: true });
       }
 
       const cleanup = () => {
+        if (cleaned) return;
+        cleaned = true;
         this._activeStreams--;
         if (signal) signal.removeEventListener('abort', onAbort);
         if (this._state === SessionState.DRAINING && this._activeStreams === 0) {
           this._close();
         }
-        if (this.available) {
-          this.onAvailable?.();
-        }
+        // Always notify the pool so it can route queued work to other sessions
+        // (or grow) when this session is draining/closed and can't take more.
+        this.onAvailable?.();
       };
 
+      stream.once('close', cleanup);
+
       stream.on('error', (err) => {
         if (!settled) {
           settled = true;
@@ -206,8 +215,17 @@ export class H2Session {
     });
   }
 
-  close(): void {
-    this._close();
+  close(): Promise<void> {
+    const session = this._session;
+    if (!session || session.destroyed) {
+      this._close();
+      return Promise.resolve();
+    }
+
+    return new Promise((resolve) => {
+      session.once('close', resolve);
+      this._close();
+    });
   }
 
   private _close(): void {

src/lib/undici-fetch.ts

@@ -37,10 +37,12 @@ import { MultipartBody } from '../_shims/MultipartBody';
 import { type Fetch } from '../core';
 
 const KEEP_ALIVE_TIMEOUT_MS = 10 * 60 * 1000;
-// Bound the pool to a few TLS sessions per origin and multiplex many H2 streams
-// over each. 4 x 64 = 256 concurrent requests in flight before undici queues the rest.
-const H2_MAX_CONNECTIONS = 4;
-const H2_MAX_CONCURRENT_STREAMS = 64;
+// Bound the pool to several TLS sessions per origin and multiplex H2 streams over
+// each. The server typically advertises MAX_CONCURRENT_STREAMS=100-128; we use 128
+// per connection to match. 20 x 128 = 2560 concurrent requests in flight before
+// undici queues the rest.
+const H2_MAX_CONNECTIONS = 20;
+const H2_MAX_CONCURRENT_STREAMS = 128;
 
 // One module-scoped default dispatcher, reused across requests: a bounded HTTP/2 pool
 // with keep-alive. `allowH2` negotiates h2 over TLS via ALPN and transparently falls

tests/lib/h2-transport.test.ts

@@ -36,6 +36,7 @@ function startTestServer(
   handler: (stream: http2.ServerHttp2Stream, headers: http2.IncomingHttpHeaders) => void,
 ): Promise<{ port: number; close: () => Promise<void> }> {
   return new Promise((resolve) => {
+    const sessions = new Set<http2.ServerHttp2Session>();
     const server = http2.createSecureServer({
       key: fs.readFileSync(keyPath),
       cert: fs.readFileSync(certPath),
@@ -44,7 +45,6 @@ function startTestServer(
       stream.on('error', () => {});
       handler(stream, headers);
     });
-    const sessions = new Set<http2.ServerHttp2Session>();
     server.on('session', (session) => {
       sessions.add(session);
       session.on('error', () => {});
@@ -56,8 +56,7 @@ function startTestServer(
         port,
         close: () =>
           new Promise<void>((res) => {
-            for (const session of sessions) session.destroy();
-            sessions.clear();
+            for (const session of sessions) session.close();
             server.close(() => res());
           }),
       });
@@ -464,4 +463,17 @@ describe('normalizeBody', () => {
     const body = await resp.json();
     expect(body.echoed).toBe('buffer data');
   });
+
+  test('Readable body is buffered', async () => {
+    const { Readable } = await import('node:stream');
+    const h2Fetch = createH2Fetch({ tlsOptions: testTls });
+    const stream = Readable.from([Buffer.from('chunk-1|'), Buffer.from('chunk-2')]);
+    const resp = (await h2Fetch(`https://localhost:${server.port}/echo`, {
+      method: 'POST',
+      headers: {},
+      body: stream,
+    })) as any;
+    const body = await resp.json();
+    expect(body.echoed).toBe('chunk-1|chunk-2');
+  });
 });