Fix marketplace fetch crash loop: consume bodyless responses and cancel the request timeout

apps/server/src/services/plugin-catalog/marketplace-http.ts

@@ -1,5 +1,6 @@
 import { lookup as dnsLookup } from "node:dns";
-import { request } from "node:https";
+import type { ClientRequest, IncomingMessage } from "node:http";
+import { request, type RequestOptions } from "node:https";
 import { BlockList, isIP, type LookupFunction } from "node:net";
 import { Readable } from "node:stream";
 
@@ -143,52 +144,91 @@ export function createPublicMarketplaceLookup(
 
 const publicMarketplaceLookup = createPublicMarketplaceLookup();
 
-export const publicMarketplaceFetch: MarketplaceFetch = async (input, init) => {
-  const url = assertPublicMarketplaceUrl(input);
-  if (init.body !== undefined && init.body !== null) {
-    throw new Error("marketplace requests cannot contain a body");
-  }
-  return new Promise<Response>((resolve, reject) => {
-    const headers = Object.fromEntries(new Headers(init.headers).entries());
-    const requestSignal = init.signal ?? undefined;
-    const outgoing = request(
-      url,
-      {
-        method: init.method ?? "GET",
-        headers,
-        lookup: publicMarketplaceLookup,
-        ...(requestSignal === undefined ? {} : { signal: requestSignal }),
-      },
-      (incoming) => {
-        const status = incoming.statusCode;
-        if (status === undefined) {
-          incoming.destroy();
-          reject(new Error("marketplace response has no HTTP status"));
-          return;
-        }
-        const responseHeaders = new Headers();
-        for (let index = 0; index < incoming.rawHeaders.length; index += 2) {
-          const name = incoming.rawHeaders[index];
-          const value = incoming.rawHeaders[index + 1];
-          if (name !== undefined && value !== undefined) {
-            responseHeaders.append(name, value);
+type MarketplaceRequest = (
+  url: URL,
+  options: RequestOptions,
+  callback: (response: IncomingMessage) => void,
+) => ClientRequest;
+
+export interface PublicMarketplaceFetchDeps {
+  request?: MarketplaceRequest;
+  lookup?: LookupFunction;
+}
+
+export function createPublicMarketplaceFetch(
+  deps: PublicMarketplaceFetchDeps = {},
+): MarketplaceFetch {
+  const sendRequest = deps.request ?? request;
+  const lookup = deps.lookup ?? publicMarketplaceLookup;
+  return async (input, init) => {
+    const url = assertPublicMarketplaceUrl(input);
+    if (init.body !== undefined && init.body !== null) {
+      throw new Error("marketplace requests cannot contain a body");
+    }
+    const timeoutController = new AbortController();
+    const timeout = setTimeout(() => {
+      timeoutController.abort(
+        new DOMException(
+          "The operation was aborted due to timeout",
+          "TimeoutError",
+        ),
+      );
+    }, MARKETPLACE_FETCH_TIMEOUT_MS);
+    const stopTimeout = () => clearTimeout(timeout);
+    const callerSignal = init.signal ?? null;
+    const requestSignal =
+      callerSignal === null
+        ? timeoutController.signal
+        : AbortSignal.any([callerSignal, timeoutController.signal]);
+    return new Promise<Response>((resolve, reject) => {
+      const headers = Object.fromEntries(new Headers(init.headers).entries());
+      const outgoing = sendRequest(
+        url,
+        {
+          method: init.method ?? "GET",
+          headers,
+          lookup,
+          signal: requestSignal,
+        },
+        (incoming) => {
+          incoming.once("close", stopTimeout);
+          const status = incoming.statusCode;
+          if (status === undefined) {
+            incoming.destroy();
+            reject(new Error("marketplace response has no HTTP status"));
+            return;
           }
-        }
-        const hasNoBody = status === 204 || status === 205 || status === 304;
-        resolve(
-          new Response(
-            hasNoBody
-              ? null
-              : (Readable.toWeb(incoming) as ReadableStream<Uint8Array>),
-            { status, headers: responseHeaders },
-          ),
-        );
-      },
-    );
-    outgoing.once("error", reject);
-    outgoing.end();
-  });
-};
+          const responseHeaders = new Headers();
+          for (let index = 0; index < incoming.rawHeaders.length; index += 2) {
+            const name = incoming.rawHeaders[index];
+            const value = incoming.rawHeaders[index + 1];
+            if (name !== undefined && value !== undefined) {
+              responseHeaders.append(name, value);
+            }
+          }
+          const hasNoBody = status === 204 || status === 205 || status === 304;
+          if (hasNoBody) incoming.resume();
+          resolve(
+            new Response(
+              hasNoBody
+                ? null
+                : (Readable.toWeb(incoming) as ReadableStream<Uint8Array>),
+              { status, headers: responseHeaders },
+            ),
+          );
+        },
+      );
+      outgoing.once("error", (error) => {
+        stopTimeout();
+        reject(error);
+      });
+      outgoing.end();
+    });
+  };
+}
+
+export const publicMarketplaceFetch: MarketplaceFetch =
+  createPublicMarketplaceFetch();
 
 export function marketplaceErrorMessage(error: unknown): string {
   if (error instanceof DOMException && error.name === "TimeoutError") {

apps/server/src/services/plugin-catalog/marketplace-icons.ts

@@ -10,7 +10,6 @@ import {
   assertPublicMarketplaceUrl,
   boundedResponseBytes,
   marketplaceErrorMessage,
-  MARKETPLACE_FETCH_TIMEOUT_MS,
   type MarketplaceFetch,
 } from "./marketplace-http.js";
 import { brandingAssetHash } from "../plugins/app-bundle.js";
@@ -283,7 +282,6 @@ async function fetchOneIcon(args: {
     method: "GET",
     headers,
     redirect: "error",
-    signal: AbortSignal.timeout(MARKETPLACE_FETCH_TIMEOUT_MS),
   });
   if (response.status === 304 && unchangedUrl) {
     await response.body?.cancel();

apps/server/src/services/plugin-catalog/marketplace-source.ts

@@ -9,7 +9,6 @@ import {
 } from "../plugins/install-sources.js";
 import {
   boundedResponseBytes,
-  MARKETPLACE_FETCH_TIMEOUT_MS,
   type MarketplaceFetch,
 } from "./marketplace-http.js";
 import {
@@ -199,7 +198,6 @@ async function materializeHttps(
       method: "GET",
       headers,
       redirect: "error",
-      signal: AbortSignal.timeout(MARKETPLACE_FETCH_TIMEOUT_MS),
     });
   }
 

apps/server/src/services/plugin-catalog/marketplace-stats.ts

@@ -2,7 +2,6 @@ import { z } from "zod";
 import { parseJsonDocument } from "../plugins/collection-manifest.js";
 import {
   boundedResponseBytes,
-  MARKETPLACE_FETCH_TIMEOUT_MS,
   type MarketplaceFetch,
 } from "./marketplace-http.js";
 
@@ -74,7 +73,6 @@ export async function fetchMarketplaceStats(args: {
     method: "GET",
     headers: new Headers({ accept: "application/json" }),
     redirect: "error",
-    signal: AbortSignal.timeout(MARKETPLACE_FETCH_TIMEOUT_MS),
   });
   if (response.status === 404) {
     await response.body?.cancel();

apps/server/src/services/plugins/managed-plugin-artifacts.ts

@@ -26,7 +26,6 @@ import {
   assertPublicMarketplaceUrl,
   boundedResponseJson,
   publicMarketplaceFetch,
-  MARKETPLACE_FETCH_TIMEOUT_MS,
   MARKETPLACE_PACKUMENT_MAX_BYTES,
 } from "../plugin-catalog/marketplace-http.js";
 import { installGitDependencies } from "./git-plugin-dependencies.js";
@@ -178,7 +177,6 @@ export function createListedRegistryNpmResolverRun(listedRegistry: string) {
       return publicMarketplaceFetch(input, {
         ...init,
         redirect: "error",
-        signal: AbortSignal.timeout(MARKETPLACE_FETCH_TIMEOUT_MS),
       });
     },
     readJson: (response) =>

apps/server/test/services/plugin-catalog/marketplace-http.test.ts

@@ -1,9 +1,12 @@
 import type { LookupAddress, LookupOptions } from "node:dns";
+import { createServer, request as httpRequest } from "node:http";
+import type { AddressInfo } from "node:net";
 import { describe, expect, it } from "vitest";
 import {
   assertPublicMarketplaceAddress,
   assertPublicMarketplaceUrl,
   boundedResponseJson,
+  createPublicMarketplaceFetch,
   createPublicMarketplaceLookup,
 } from "../../../src/services/plugin-catalog/marketplace-http.js";
 
@@ -80,4 +83,45 @@ describe("marketplace HTTP policy", () => {
       boundedResponseJson(response, 1024, "npm registry metadata"),
     ).rejects.toThrow(/exceeds 1024 bytes/u);
   });
+
+  it("settles a bodyless response so a late timeout cannot abort the socket", async () => {
+    const server = createServer((_request, response) => {
+      response.writeHead(304);
+      response.end();
+    });
+    await new Promise<void>((resolve) =>
+      server.listen(0, "127.0.0.1", resolve),
+    );
+    const port = (server.address() as AddressInfo).port;
+    const socketErrors: Error[] = [];
+    const fetchMarketplace = createPublicMarketplaceFetch({
+      request: (url, options, callback) => {
+        const outgoing = httpRequest(
+          { ...options, host: "127.0.0.1", port, path: url.pathname },
+          callback,
+        );
+        outgoing.on("socket", (socket) => {
+          socket.on("error", (error: Error) => socketErrors.push(error));
+        });
+        return outgoing;
+      },
+    });
+    try {
+      const response = await fetchMarketplace(
+        "https://marketplace.test/marketplace.json",
+        {
+          method: "GET",
+          headers: new Headers({ accept: "application/json" }),
+          signal: AbortSignal.timeout(50),
+        },
+      );
+      expect(response.status).toBe(304);
+      await response.body?.cancel();
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      expect(socketErrors).toEqual([]);
+    } finally {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  });
 });