Lower fetch members typed by @types/node, and run the poll phase once per loop iteration (#1435)

packages/compiler/src/frontend/lowering/lower-calls.ts

@@ -3677,8 +3677,20 @@ function lowerDynReceiverMethodCall(
         )
     )
       return true;
+    // A standard-library member spelled as a function-typed property
+    // (undici-types declares Headers.append/forEach/... and the body mixin
+    // as `readonly append: (...) => void`; scriptc's own declarations use
+    // methods) is still the handle's native method: it takes the shared
+    // receiver dispatch below either way (which also calls a plain object's
+    // own member), so the lowering does not depend on which declaration
+    // file supplies the type.
+    const libMethod =
+      prop !== undefined &&
+      prop.declarations.length > 0 &&
+      prop.declarations.every((decl) => lowerer.isStdlibFile(decl.getSourceFile()));
     return (
       prop !== undefined &&
+      !libMethod &&
       prop.declarations.some(
         (decl) => ts.isPropertyDeclaration(decl) || ts.isPropertySignature(decl),
       )

packages/compiler/src/frontend/lowering/lower-stdlib.ts

@@ -621,6 +621,20 @@ export function lowerFetchCall(lowerer: Lowerer, call: ts.CallExpression): IrExp
         : lowerer.lowerExpr(argument);
       stmts.push({ kind: "exprStmt", expr: discarded, loc: discarded.loc });
     }
+    // A union input holding a URL arm (`string | URL`) converts through the
+    // checked-dynamic input path, which reads the URL's href at the call
+    // like the bare-URL arm below; narrowing it to a string would throw on
+    // the URL arm.
+    const unionWithUrl =
+      input.type.kind === "union" &&
+      (lowerer.unions.get(input.type.unionId)?.arms.some((arm) => arm.kind === "url") ?? false) &&
+      lowerer.dynConvertible(input.type);
+    const dynInput: IrExpr | null =
+      input.type.kind === "dyn"
+        ? inputRef
+        : unionWithUrl
+          ? { kind: "dynFrom", value: inputRef, type: DYN, loc: locOf(inputNode) }
+          : null;
     const url: IrExpr =
       input.type.kind === "url"
         ? {
@@ -630,13 +644,13 @@ export function lowerFetchCall(lowerer: Lowerer, call: ts.CallExpression): IrExp
             type: STRING,
             loc: locOf(inputNode),
           }
-        : input.type.kind === "dyn"
-          ? inputRef
+        : dynInput !== null
+          ? dynInput
           : lowerer.coerceInto(inputNode, inputRef, STRING);
     const answer: IrExpr = {
       kind: "libCall",
-      fn: input.type.kind === "dyn" ? "fetch.input" : "fetch.start",
-      args: [input.type.kind === "dyn" ? inputRef : url, initRef],
+      fn: dynInput !== null ? "fetch.input" : "fetch.start",
+      args: [url, initRef],
       type: { kind: "promise", inner: DYN },
       loc,
     };

packages/compiler/src/frontend/type-mapper.ts

@@ -4321,7 +4321,9 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
       // semantics would DEGRADE under the collapse — class instances,
       // Maps/Sets, functions (closure identity, `x === String`
       // narrowing), promises, regexes, generators, handles — keep their
-      // existing union homes and fences below. A pending recursive
+      // existing union homes and fences below. URL is the
+      // exception: its handle boxes by reference and keeps every member,
+      // so it collapses too (dynSubsumableUnionArm). A pending recursive
       // placeholder cannot collapse (back-references already hold the
       // union id); the degenerate recursive spelling keeps its fence.
       if (
@@ -6680,6 +6682,13 @@ function dynSubsumableUnionArm(arm: IrType, ctx: TypeMapperCtx): boolean {
     case "undefinedT":
     case "nullT":
       return true;
+    // URL boxes BY REFERENCE (its checked-dynamic handle keeps identity,
+    // reads and writes its members through the native accessors, answers
+    // `instanceof URL`, and checks back out to the typed value), so it
+    // rides the collapse beside fetch's Request/Response handles:
+    // `string | URL | Request`, fetch's own input type under @types/node.
+    case "url":
+      return true;
     case "record":
     case "array":
       return canConvertToDyn(

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6251-net-connect-poll-phase-order.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6251-net-connect-poll-phase-order.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6252-fetch-input-url-unions.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6252-fetch-input-url-unions.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/fetch-handle-members.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/fixtures/node-types/fetch-handle-members.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/fetch-input-unions.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/fixtures/node-types/fetch-input-unions.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_async.c

@@ -4138,6 +4138,25 @@ static int scr_workers_pollfd(void) {
  * macrotask). */
 bool scr_loop_has_ready(void) { return scr_ready_len > 0 || scr_nt_head != NULL; }
 
+/* The poll phase runs once per loop iteration, like libuv's: the net unit
+ * drains the poller at most once between check phases. I/O that becomes
+ * ready while this iteration's callbacks run (a loopback connect started
+ * by one of them) waits for the next iteration, so immediates queued in
+ * the meantime fire first — Node's poll → check order. A full batch polls
+ * again without waiting, at most SCR_LOOP_IO_BATCHES times per phase (the
+ * count libuv's uv__io_poll uses); later events wait for the next
+ * iteration, so timers and immediates still run while sockets stay busy. */
+enum { SCR_LOOP_IO_BATCHES = 48 };
+static SCR_TL bool scr_loop_io_open = true;
+static SCR_TL int scr_loop_io_batches = 0;
+bool scr_loop_io_take(void) {
+  bool open = scr_loop_io_open;
+  scr_loop_io_open = false;
+  return open;
+}
+bool scr_loop_io_more(void) { return ++scr_loop_io_batches < SCR_LOOP_IO_BATCHES; }
+void scr_loop_io_reopen(void) { scr_loop_io_open = true; }
+
 /* The process.nextTick queue to exhaustion, ticks queued meanwhile
  * included. false: a tick threw (uncaught, pending). */
 static bool scr_loop_run_ticks(void) {
@@ -4615,6 +4634,7 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) {
      * every macrotask; pending ticks or a woken fiber restart the turn
      * first (the timer loop's break above), so the phase only runs on a
      * drained checkpoint. */
+    bool check_ran = false;
     if (scr_ready_len == 0 && scr_nt_head == NULL && scr_nunhandled == 0 && scr_pending_immediates > 0) {
       size_t end = scr_nimmediates;
       while (scr_immediates_head < end) {
@@ -4645,6 +4665,15 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) {
         scr_immediates_head = 0;
         scr_nimmediates = 0;
       }
+      check_ran = true;
+    }
+    /* The next poll phase opens once this iteration's check phase ran (or
+     * had nothing to run). A check phase deferred for pending microtasks
+     * keeps it closed, so the next iteration reaches the check phase
+     * before polling again. */
+    if (check_ran || scr_pending_immediates == 0) {
+      scr_loop_io_open = true;
+      scr_loop_io_batches = 0;
     }
   }
   return rejection_failed;

packages/runtime/src/scr_net.c

@@ -4908,17 +4908,30 @@ static int scr_net_pollfd(void) {
   return scr_net_poller != NULL ? scrp_poller_fd(scr_net_poller) : -1;
 }
 
+bool scr_loop_io_take(void);
+bool scr_loop_io_more(void);
+void scr_loop_io_reopen(void);
+
 static void scr_net_dispatch(void) {
   if (!scr_net_servers && !scr_net_socks &&
       !scr_net_proto_pending()) {
     return;
   }
+  /* One poller drain per loop iteration (scr_loop_io_take); a full batch
+   * keeps draining within the phase's batch budget (scr_loop_io_more), like
+   * libuv's poll phase. Deferred emits (the sweep) still run on every
+   * call. */
+  bool may_drain = scr_loop_io_take();
   for (;;) {
     scr_net_sweep();
     if (scr_exc_pending()) return;
     if (scr_net_poller == NULL) return;
     ScrPollerEvent evs[64];
-    int n = scrp_drain(scr_net_poller, evs, 64);
+    int n = 0;
+    if (may_drain) {
+      n = scrp_drain(scr_net_poller, evs, 64);
+      may_drain = n == 64 && scr_loop_io_more();
+    }
     for (int i = 0; i < n; i++) {
       void *udata = evs[i].udata;
       if (!udata) continue;
@@ -4992,8 +5005,13 @@ static void scr_net_dispatch(void) {
       }
       if (scr_exc_pending()) return;
     }
-    if (!scr_net_flags_pending()) return;
-    if (scr_loop_has_ready()) return; /* microtasks interleave first */
+    if (!scr_net_flags_pending() && !may_drain) return;
+    if (scr_loop_has_ready()) {
+      /* microtasks interleave first; a full batch's rest drains on the
+       * next call, still within this poll phase */
+      if (may_drain) scr_loop_io_reopen();
+      return;
+    }
   }
 }
 

tests/corpus/6251-net-connect-poll-phase-order.ts

@@ -0,0 +1,70 @@
+// A connection started by a poll-phase callback completes in the NEXT loop
+// iteration's poll phase: the callback's nextTick and promise jobs run
+// first, then the check phase's setImmediate, then the new socket's
+// 'connect' (libuv/Node order; loopback connects complete at once, so an
+// event loop that polls again before the check phase reorders them). The
+// second half is the same through an awaited promise, the shape of an HTTP
+// client that dials a fresh connection after the previous response.
+import * as net from "node:net";
+
+const log: string[] = [];
+const server = net.createServer((socket) => {
+  socket.on("data", (chunk: Buffer) => socket.write(chunk));
+});
+
+function dial(port: number, label: string): Promise<net.Socket> {
+  return new Promise((resolve) => {
+    const socket = net.connect({ port, host: "127.0.0.1" });
+    socket.on("connect", () => {
+      log.push(`connect ${label}`);
+      resolve(socket);
+    });
+  });
+}
+
+function roundTrip(socket: net.Socket, label: string): Promise<void> {
+  return new Promise((resolve) => {
+    socket.once("data", () => {
+      log.push(`data ${label}`);
+      setImmediate(() => log.push(`immediate after data ${label}`));
+      process.nextTick(() => log.push(`tick after data ${label}`));
+      Promise.resolve().then(() => log.push(`promise after data ${label}`));
+      resolve();
+    });
+    socket.write("ping");
+  });
+}
+
+server.listen(0, "127.0.0.1", () => {
+  const address = server.address();
+  const port = address !== null && typeof address === "object" ? address.port : 0;
+  const first = net.connect({ port, host: "127.0.0.1" });
+  first.on("connect", () => {
+    log.push("connect first");
+    first.write("ping");
+  });
+  first.once("data", () => {
+    log.push("data first");
+    setImmediate(() => log.push("immediate from data"));
+    process.nextTick(() => log.push("tick from data"));
+    Promise.resolve().then(() => log.push("promise from data"));
+    const second = net.connect({ port, host: "127.0.0.1" });
+    second.on("connect", async () => {
+      log.push("connect second");
+      second.destroy();
+      first.destroy();
+      // Awaited: each response's continuation dials the next connection.
+      const a = await dial(port, "a");
+      await roundTrip(a, "a");
+      log.push("continuation after a");
+      const b = await dial(port, "b");
+      await roundTrip(b, "b");
+      a.destroy();
+      b.destroy();
+      setImmediate(() => {
+        server.close();
+        console.log(log.join("\n"));
+      });
+    });
+  });
+});

tests/corpus/6252-fetch-input-url-unions.ts

@@ -0,0 +1,48 @@
+// `typeof globalThis.fetch` under @types/node takes `string | URL | Request`:
+// unions holding URL handles beside fetch handles compile (URL boxes by
+// reference into the checked-dynamic representation), and a custom fetch
+// cast to that type through `unknown` keeps working, including an extra
+// init member written through a record view (the dispatcher pattern).
+import * as http from "node:http";
+type FetchImpl = typeof globalThis.fetch;
+interface DispatchingInit {
+  method?: string;
+  headers?: Record<string, string>;
+  dispatcher?: { name: string };
+}
+function customFetch(input: string | URL, init?: DispatchingInit): Promise<Response> {
+  console.log("custom fetch", typeof input === "string" ? "string" : "url", init?.dispatcher?.name ?? "none");
+  return fetch(input, { method: init?.method, headers: init?.headers });
+}
+const server = http.createServer((req, res) => res.end(`ok ${req.method} ${req.url}`));
+server.listen(0, "127.0.0.1", async () => {
+  const address = server.address();
+  const port = address !== null && typeof address === "object" ? address.port : 0;
+  const impl: FetchImpl = customFetch as unknown as FetchImpl;
+  const init: RequestInit = { method: "POST" };
+  (init as Record<string, unknown>).dispatcher = { name: "pinned" };
+  const res = await impl(`http://127.0.0.1:${port}/a`, init);
+  console.log(res.status, await res.text());
+  const viaUrl = await impl(new URL(`http://127.0.0.1:${port}/b`));
+  console.log(viaUrl.status, await viaUrl.text());
+  const direct = await fetch(new URL(`http://127.0.0.1:${port}/c`));
+  console.log(direct.status, await direct.text());
+  server.close();
+});
+
+function describeInput(input: string | Request | URL): string {
+  if (typeof input === "string") return `string ${input}`;
+  if (input instanceof URL) return `url ${input.href} ${input.hostname}`;
+  return `request ${input.method} ${input.url}`;
+}
+console.log(describeInput("http://a.test/x"));
+const u = new URL("http://b.test:8080/p?q=1");
+console.log(describeInput(u));
+console.log(describeInput(new Request("http://c.test/r", { method: "PUT" })));
+const items: (string | URL | Request)[] = [u, "s"];
+console.log(items.length, items[0] === u, String(items[0]));
+let slot: string | URL | Request = u;
+if (slot instanceof URL) { slot.pathname = "/changed"; }
+console.log(u.href);
+const params: string | Request | URL = new URL("http://d.test/?b=2");
+console.log(typeof params === "string" ? params : params instanceof URL ? params.searchParams.get("b") : "req");

tests/fixtures/node-types/fetch-handle-members.ts

@@ -0,0 +1,32 @@
+// undici-types (@types/node's fetch declarations) spells the Headers methods
+// and the Request/Response body mixin as function-typed properties; they
+// must lower to the same native handle methods as the shipped declarations.
+async function main(): Promise<void> {
+  const h = new Headers([["b", "2"], ["a", "1"]]);
+  h.append("set-cookie", "x=1");
+  h.append("set-cookie", "y=2");
+  h.set("c", "3");
+  h.delete("b");
+  const seen: string[] = [];
+  h.forEach((value, name) => seen.push(`${name}=${value}`));
+  console.log(seen.join(","), h.get("a"), h.has("b"), h.getSetCookie().join("|"));
+  const r1 = new Response("body-text", { status: 202, statusText: "Acc", headers: h });
+  console.log(r1.status, r1.statusText, r1.ok, r1.headers.get("c"), r1.bodyUsed);
+  console.log(await r1.text(), r1.bodyUsed);
+  const r2 = new Response(JSON.stringify({ k: 1 }));
+  console.log(JSON.stringify(await r2.json()));
+  const r3 = new Response(new Uint8Array([1, 2, 3]));
+  console.log((await r3.arrayBuffer()).byteLength);
+  const r4 = new Response("abc");
+  console.log((await r4.bytes()).length);
+  const r5 = Response.json({ z: true }, { status: 200 });
+  console.log(r5.headers.get("content-type"), await r5.text());
+  const req = new Request("http://x.test/a?b=1", { method: "POST", body: "payload", headers: { q: "1" } });
+  console.log(req.method, req.url, req.headers.get("q"), await req.text());
+  const req2 = new Request("http://x.test/j", { method: "PUT", body: "{\"n\":2}" });
+  console.log(JSON.stringify(await req2.json()));
+  const ac = new AbortController();
+  ac.abort();
+  console.log(ac.signal.aborted);
+}
+main();

tests/fixtures/node-types/fetch-input-unions.ts

@@ -0,0 +1,48 @@
+// `typeof globalThis.fetch` under @types/node takes `string | URL | Request`:
+// unions holding URL handles beside fetch handles compile (URL boxes by
+// reference into the checked-dynamic representation), and a custom fetch
+// cast to that type through `unknown` keeps working, including an extra
+// init member written through a record view (the dispatcher pattern).
+import * as http from "node:http";
+type FetchImpl = typeof globalThis.fetch;
+interface DispatchingInit {
+  method?: string;
+  headers?: Record<string, string>;
+  dispatcher?: { name: string };
+}
+function customFetch(input: string | URL, init?: DispatchingInit): Promise<Response> {
+  console.log("custom fetch", typeof input === "string" ? "string" : "url", init?.dispatcher?.name ?? "none");
+  return fetch(input, { method: init?.method, headers: init?.headers });
+}
+const server = http.createServer((req, res) => res.end(`ok ${req.method} ${req.url}`));
+server.listen(0, "127.0.0.1", async () => {
+  const address = server.address();
+  const port = address !== null && typeof address === "object" ? address.port : 0;
+  const impl: FetchImpl = customFetch as unknown as FetchImpl;
+  const init: RequestInit = { method: "POST" };
+  (init as Record<string, unknown>).dispatcher = { name: "pinned" };
+  const res = await impl(`http://127.0.0.1:${port}/a`, init);
+  console.log(res.status, await res.text());
+  const viaUrl = await impl(new URL(`http://127.0.0.1:${port}/b`));
+  console.log(viaUrl.status, await viaUrl.text());
+  const direct = await fetch(new URL(`http://127.0.0.1:${port}/c`));
+  console.log(direct.status, await direct.text());
+  server.close();
+});
+
+function describeInput(input: string | Request | URL): string {
+  if (typeof input === "string") return `string ${input}`;
+  if (input instanceof URL) return `url ${input.href} ${input.hostname}`;
+  return `request ${input.method} ${input.url}`;
+}
+console.log(describeInput("http://a.test/x"));
+const u = new URL("http://b.test:8080/p?q=1");
+console.log(describeInput(u));
+console.log(describeInput(new Request("http://c.test/r", { method: "PUT" })));
+const items: (string | URL | Request)[] = [u, "s"];
+console.log(items.length, items[0] === u, String(items[0]));
+let slot: string | URL | Request = u;
+if (slot instanceof URL) { slot.pathname = "/changed"; }
+console.log(u.href);
+const params: string | Request | URL = new URL("http://d.test/?b=2");
+console.log(typeof params === "string" ? params : params instanceof URL ? params.searchParams.get("b") : "req");

tests/fixtures/npm/cases/undici5/main.ts

@@ -196,7 +196,6 @@ server.listen(0, "127.0.0.1", async () => {
   for (let i = 0; i < 4; i++) {
     const r = await request(`${origin}/r${i}`, { dispatcher: agent });
     await r.body.text();
-    await turn();
   }
   console.log("connections opened:", agent.connections);
   console.log("stats:", agent.statsFor(origin));

tests/fixtures/npm/cases/undici7/main.ts

@@ -202,7 +202,6 @@ server.listen(0, "127.0.0.1", async () => {
   for (let i = 0; i < 4; i++) {
     const r = await request(`${origin}/r${i}`, { dispatcher: agent });
     await r.body.text();
-    await turn();
   }
   console.log("connections opened:", agent.connections);
   console.log("stats:", agent.statsFor(origin));

tests/harness/project-config.test.ts

@@ -52,6 +52,48 @@ test("node-types: filesystem capacity supports typed number and BigInt results u
   expect(comparableStderr(native.stderr)).toBe(node.stderr);
 });
 
+test("node-types: Headers, Request and Response members typed by undici-types run natively", async () => {
+  const outDir = outDirFor("fetch-handle-members");
+  const entry = join(nodeTypesDir, "fetch-handle-members.ts");
+  const result = await compile(entry, {
+    outPath: join(
+      outDir,
+      process.platform === "win32" ? "fetch-handle-members.exe" : "fetch-handle-members",
+    ),
+    outDir,
+    sanitize,
+  });
+  expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
+  if (!result.ok) return;
+  const [native, node] = await Promise.all([
+    execFileAsync(result.binaryPath),
+    execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
+  ]);
+  expect(native.stdout).toBe(node.stdout);
+  expect(comparableStderr(native.stderr)).toBe(node.stderr);
+});
+
+test("node-types: fetch's string | URL | Request input and a custom fetch cast to typeof fetch", async () => {
+  const outDir = outDirFor("fetch-input-unions");
+  const entry = join(nodeTypesDir, "fetch-input-unions.ts");
+  const result = await compile(entry, {
+    outPath: join(
+      outDir,
+      process.platform === "win32" ? "fetch-input-unions.exe" : "fetch-input-unions",
+    ),
+    outDir,
+    sanitize,
+  });
+  expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
+  if (!result.ok) return;
+  const [native, node] = await Promise.all([
+    execFileAsync(result.binaryPath),
+    execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
+  ]);
+  expect(native.stdout).toBe(node.stdout);
+  expect(comparableStderr(native.stderr)).toBe(node.stderr);
+});
+
 test("node-types: http header arrays and server.address() follow @types/node's types", async () => {
   const outDir = outDirFor("http-headers-address");
   const entry = join(nodeTypesDir, "http-headers-address.ts");