diff --git a/packages/runtime-playground/src/preview-server.ts b/packages/runtime-playground/src/preview-server.ts index 299b29d5..e34982dc 100644 --- a/packages/runtime-playground/src/preview-server.ts +++ b/packages/runtime-playground/src/preview-server.ts @@ -136,7 +136,18 @@ function previewProxyServer(target: URL, routes: InternalPreviewRouteRegistry): return } - upstreamQueue(() => proxyPreviewRequest(target, incoming, outgoing)).catch((error: Error) => writeProxyError(outgoing, error)) + upstreamQueue( + () => proxyPreviewRequest(target, incoming, outgoing), + () => incoming.aborted || incoming.destroyed || outgoing.destroyed, + (cancel) => { + incoming.once("aborted", cancel) + outgoing.once("close", cancel) + return () => { + incoming.off("aborted", cancel) + outgoing.off("close", cancel) + } + }, + ).catch((error: Error) => writeProxyError(outgoing, error)) }) } @@ -203,6 +214,8 @@ function proxyPreviewRequest(target: URL, incoming: IncomingMessage, outgoing: S outgoing.destroy(error) settle() }) + response.on("end", settle) + response.on("close", settle) outgoing.on("finish", settle) outgoing.on("close", settle) if (bodyTransform) { @@ -218,6 +231,7 @@ function proxyPreviewRequest(target: URL, incoming: IncomingMessage, outgoing: S writeProxyError(outgoing, error) settle() }) + incoming.on("aborted", abortUpstream) incoming.on("error", () => { abortUpstream() }) @@ -258,37 +272,80 @@ function previewProxyRequestTarget(incoming: IncomingMessage, target: URL): Prev return { upstreamHost: target.host, visibleHost: authority.host, path: rawUrl, port: authority.port || "80", protocol: "http:", rewriteTargetOrigin: true } } -function createPreviewProxyQueue(): (task: () => Promise) => Promise { +function createPreviewProxyQueue(): ( + task: () => Promise, + isCanceled: () => boolean, + observeCancellation: (cancel: () => void) => () => void, +) => Promise { let active = false - const pending: Array<() => void> = [] - - const acquire = async () => { - if (!active) { - active = true - return - } - - await new Promise((resolve) => pending.push(resolve)) + interface PendingRequest { + task: () => Promise + isCanceled: () => boolean + stopObservingCancellation: () => void + resolve: () => void + reject: (error: unknown) => void } - - const release = () => { - const next = pending.shift() - if (next) { - next() - return + const pending: PendingRequest[] = [] + + function release(): void { + let next = pending.shift() + while (next) { + next.stopObservingCancellation() + if (!next.isCanceled()) { + run(next) + return + } + next.resolve() + next = pending.shift() } - active = false } - return async (task) => { - await acquire() + function run(request: PendingRequest): void { + request.stopObservingCancellation() + if (request.isCanceled()) { + request.resolve() + release() + return + } + let task: Promise try { - await task() - } finally { + task = request.task() + } catch (error) { + request.reject(error) release() + return } + task.then(request.resolve, request.reject).finally(release) } + + return (task, isCanceled, observeCancellation) => new Promise((resolve, reject) => { + const request: PendingRequest = { + task, + isCanceled, + stopObservingCancellation: () => {}, + resolve, + reject, + } + const cancel = () => { + const index = pending.indexOf(request) + if (index === -1) { + return + } + pending.splice(index, 1) + request.stopObservingCancellation() + resolve() + } + request.stopObservingCancellation = observeCancellation(cancel) + + if (active) { + pending.push(request) + return + } + + active = true + run(request) + }) } async function listenPreviewProxy(proxy: PreviewProxyServer, port: number, bind: string): Promise { diff --git a/tests/browser-callback-materialization-contracts.test.ts b/tests/browser-callback-materialization-contracts.test.ts index 05bb86f3..d5daf8bb 100644 --- a/tests/browser-callback-materialization-contracts.test.ts +++ b/tests/browser-callback-materialization-contracts.test.ts @@ -259,6 +259,65 @@ try { await closeHttpServer(hangingTargetServer) } +let releaseFirstQueuedRequest: (() => void) | undefined +let firstQueuedRequestReached: (() => void) | undefined +const firstQueuedRequest = new Promise((resolve) => { + firstQueuedRequestReached = resolve +}) +const queuedTargetRequests: string[] = [] +const queuedTargetServer = createServer((request, response) => { + queuedTargetRequests.push(request.url ?? "") + if (request.url === "/first") { + firstQueuedRequestReached?.() + new Promise((resolve) => { + releaseFirstQueuedRequest = resolve + }).then(() => response.end("first")) + return + } + response.end("third") +}) +const queuedTargetServerUrl = await listenLocalHttpServer(queuedTargetServer) +const queuedAbortProxy = await withPreviewProxy({ + playground: { async run() { return { text: "" } } }, + serverUrl: queuedTargetServerUrl, + async [Symbol.asyncDispose]() {}, +} satisfies PlaygroundCliServer, 0) +let secondQueuedRequestReached: (() => void) | undefined +const secondQueuedRequest = new Promise((resolve) => { + secondQueuedRequestReached = resolve +}) +let secondQueuedRequestCanceled: (() => void) | undefined +const secondQueuedCancellation = new Promise((resolve) => { + secondQueuedRequestCanceled = resolve +}) +const stopObservingQueuedRequest = queuedAbortProxy.previewRoutes?.add((request, response) => { + if (request.url === "/second") { + secondQueuedRequestReached?.() + response.once("close", () => secondQueuedRequestCanceled?.()) + } + return false +}) +try { + const firstRequest = fetch(`${queuedAbortProxy.serverUrl}/first`) + await firstQueuedRequest + + const controller = new AbortController() + const canceledRequest = fetch(`${queuedAbortProxy.serverUrl}/second`, { signal: controller.signal }).catch((error) => error) + await secondQueuedRequest + controller.abort() + await canceledRequest + await secondQueuedCancellation + + releaseFirstQueuedRequest?.() + assert.equal(await (await firstRequest).text(), "first") + assert.equal(await (await fetch(`${queuedAbortProxy.serverUrl}/third`)).text(), "third") + assert.deepEqual(queuedTargetRequests, ["/first", "/third"], "a canceled queued request must not open an upstream request") +} finally { + stopObservingQueuedRequest?.() + await queuedAbortProxy[Symbol.asyncDispose]() + await closeHttpServer(queuedTargetServer) +} + const phase = materializationPhaseResult({ phase: "persist-browser-artifacts", status: "completed",