From 66b160b0e2e15758fad7f7ac51de16f94fa1c583 Mon Sep 17 00:00:00 2001 From: Alejo Thomas Ortega Date: Wed, 7 Jan 2026 11:57:06 -0300 Subject: [PATCH 1/3] fix: gracefully close connections on errors while piping responses --- src/logic.ts | 3 +++ test/integration.spec.ts | 32 ++++++++++++++++++++++++++++++-- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/src/logic.ts b/src/logic.ts index e27d883..0f0ff75 100644 --- a/src/logic.ts +++ b/src/logic.ts @@ -70,6 +70,9 @@ export function success(data: fetch.Response, res: http.ServerResponse) { // } throw new Error('Unknown response body (Blob)') } else if (body && body.pipe) { + body.on('error', (err) => { + res.destroy(err) + }) body.pipe(res) // Note: for context about why this is necessary, check https://github.com/nodejs/node/issues/1180 diff --git a/test/integration.spec.ts b/test/integration.spec.ts index b8cba92..fc02b23 100644 --- a/test/integration.spec.ts +++ b/test/integration.spec.ts @@ -12,7 +12,7 @@ import { multipartParserWrapper } from './busboy' describeE2E('integration sanity tests using http backend', integrationSuite) describeTestE2E('integration sanity tests using test server', integrationSuite) -describeTestE2E('underlying server', function({ components }: { components: TestComponents }) { +describeTestE2E('underlying server', function ({ components }: { components: TestComponents }) { it('gets the underlying http server', async () => { const { server } = components const http = getUnderlyingServer(server) @@ -416,7 +416,7 @@ function integrationSuite({ components }: { components: TestComponents }) { const results = new Set<{ id: number }>() let i = 0 server.use(async (ctx) => { - ; (ctx as any).id = i++ + ;(ctx as any).id = i++ results.add(ctx as any) return null as any }) @@ -592,5 +592,33 @@ function integrationSuite({ components }: { components: TestComponents }) { const res = await fetch.fetch(`/hola`) expect(res.status).toEqual(500) }) + + it('gracefully closes socket when an error occurs during piping', async () => { + const { fetch, server } = components + server.resetMiddlewares() + + const routes = new Router() + + routes.get('/stream-error', async (ctx) => { + function* streamContent() { + yield 'some data' + // simulate error while piping + throw new Error('Stream error during piping') + } + + return { + status: 200, + body: Stream.Readable.from(streamContent(), { encoding: 'utf-8' }) + } + }) + + server.use(routes.middleware()) + + { + const res = await fetch.fetch(`/stream-error`) + expect(res.status).toEqual(200) + await expect(res.text()).rejects.toThrow() + } + }) }) } From 0150d9c2dc05e69051360b1357d83b1344fa3d3f Mon Sep 17 00:00:00 2001 From: Alejo Thomas Ortega Date: Wed, 7 Jan 2026 17:12:27 -0300 Subject: [PATCH 2/3] chore: apply PR feedback to use .end instead of .destroy on pipe errors --- src/logic.ts | 4 ++-- test/integration.spec.ts | 48 ++++++++++++++++++++++------------------ 2 files changed, 28 insertions(+), 24 deletions(-) diff --git a/src/logic.ts b/src/logic.ts index 0f0ff75..583fd25 100644 --- a/src/logic.ts +++ b/src/logic.ts @@ -70,8 +70,8 @@ export function success(data: fetch.Response, res: http.ServerResponse) { // } throw new Error('Unknown response body (Blob)') } else if (body && body.pipe) { - body.on('error', (err) => { - res.destroy(err) + body.on('error', (_err) => { + res.end() }) body.pipe(res) diff --git a/test/integration.spec.ts b/test/integration.spec.ts index fc02b23..5617c8f 100644 --- a/test/integration.spec.ts +++ b/test/integration.spec.ts @@ -593,32 +593,36 @@ function integrationSuite({ components }: { components: TestComponents }) { expect(res.status).toEqual(500) }) - it('gracefully closes socket when an error occurs during piping', async () => { - const { fetch, server } = components - server.resetMiddlewares() - - const routes = new Router() + }) +} - routes.get('/stream-error', async (ctx) => { - function* streamContent() { - yield 'some data' - // simulate error while piping - throw new Error('Stream error during piping') - } +describeE2E('stream error handling while piping', function ({ components }: { components: TestComponents }) { + it('gracefully ends response', async () => { + const { fetch, server } = components + server.resetMiddlewares() - return { - status: 200, - body: Stream.Readable.from(streamContent(), { encoding: 'utf-8' }) - } - }) + const routes = new Router() - server.use(routes.middleware()) + routes.get('/stream-error', async (ctx) => { + function* streamContent() { + yield 'some data' + // simulate error while piping + throw new Error('Stream error during piping') + } - { - const res = await fetch.fetch(`/stream-error`) - expect(res.status).toEqual(200) - await expect(res.text()).rejects.toThrow() + return { + status: 200, + body: Stream.Readable.from(streamContent(), { encoding: 'utf-8' }) } }) + + server.use(routes.middleware()) + + { + const res = await fetch.fetch(`/stream-error`) + expect(res.status).toEqual(200) + const text = await res.text() + expect(text).toEqual('some data') + } }) -} +}) From 3e57dfb9d3c414da1627e255247cef7b10ef9c10 Mon Sep 17 00:00:00 2001 From: Alejo Thomas Ortega Date: Thu, 8 Jan 2026 10:51:39 -0300 Subject: [PATCH 3/3] chore: log errors while piping responses --- src/logic.ts | 7 ++++--- src/server.ts | 4 ++-- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/logic.ts b/src/logic.ts index 583fd25..56a9097 100644 --- a/src/logic.ts +++ b/src/logic.ts @@ -4,7 +4,7 @@ import * as http from 'http' import * as https from 'https' import destroy from 'destroy' import onFinished from 'on-finished' -import type { IHttpServerComponent } from '@well-known-components/interfaces' +import type { IHttpServerComponent, ILoggerComponent } from '@well-known-components/interfaces' import type { IHttpServerOptions } from './types' import { HttpError } from 'http-errors' import { Middleware } from './middleware' @@ -43,7 +43,7 @@ export const isBlob = (object: any): object is Blob => { /** * @internal */ -export function success(data: fetch.Response, res: http.ServerResponse) { +export function success(data: fetch.Response, res: http.ServerResponse, options: { logger: ILoggerComponent.ILogger }) { if (data.statusText) res.statusMessage = data.statusText if (data.status) res.statusCode = data.status @@ -70,7 +70,8 @@ export function success(data: fetch.Response, res: http.ServerResponse) { // } throw new Error('Unknown response body (Blob)') } else if (body && body.pipe) { - body.on('error', (_err) => { + body.on('error', (err) => { + options.logger.error('Error piping response body', { error: err.message }) res.end() }) body.pipe(res) diff --git a/src/server.ts b/src/server.ts index a311655..38b4c21 100644 --- a/src/server.ts +++ b/src/server.ts @@ -113,8 +113,8 @@ export async function createServerComponent( async function asyncHandle(req: http.IncomingMessage, res: http.ServerResponse) { const request = getRequestFromNodeMessage(req, host) const response = await serverHandler.processRequest(configuredContext, request) - - success(response, res) + + success(response, res, { logger }) } async function handleUpgrade(req: http.IncomingMessage, socket: Socket, head: Buffer) {