diff --git a/apps/server/src/services/plugin-catalog/marketplace-http.ts b/apps/server/src/services/plugin-catalog/marketplace-http.ts index fe4a29ae3e0..c321ac8cab9 100644 --- a/apps/server/src/services/plugin-catalog/marketplace-http.ts +++ b/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((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((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), - { 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), + { 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") { diff --git a/apps/server/src/services/plugin-catalog/marketplace-icons.ts b/apps/server/src/services/plugin-catalog/marketplace-icons.ts index fa912941edc..426c9852adf 100644 --- a/apps/server/src/services/plugin-catalog/marketplace-icons.ts +++ b/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(); diff --git a/apps/server/src/services/plugin-catalog/marketplace-source.ts b/apps/server/src/services/plugin-catalog/marketplace-source.ts index 1cfd10b111f..2637f3b6088 100644 --- a/apps/server/src/services/plugin-catalog/marketplace-source.ts +++ b/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), }); } diff --git a/apps/server/src/services/plugin-catalog/marketplace-stats.ts b/apps/server/src/services/plugin-catalog/marketplace-stats.ts index f8ceaccb5b9..86213fbca9d 100644 --- a/apps/server/src/services/plugin-catalog/marketplace-stats.ts +++ b/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(); diff --git a/apps/server/src/services/plugins/managed-plugin-artifacts.ts b/apps/server/src/services/plugins/managed-plugin-artifacts.ts index 73257b6661b..41636dfd400 100644 --- a/apps/server/src/services/plugins/managed-plugin-artifacts.ts +++ b/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) => diff --git a/apps/server/test/services/plugin-catalog/marketplace-http.test.ts b/apps/server/test/services/plugin-catalog/marketplace-http.test.ts index 3f8f7ea0872..e218f7405b7 100644 --- a/apps/server/test/services/plugin-catalog/marketplace-http.test.ts +++ b/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((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((resolve) => server.close(() => resolve())); + } + }); });