From 6cfb6adbd7bf470812b689ef9d21a9a721dd0f11 Mon Sep 17 00:00:00 2001 From: Ogunleye Pelumi Date: Tue, 29 Sep 2026 19:01:48 +0100 Subject: [PATCH 1/3] security: Destroy sockets when errors occur after headers flush (#1277) --- src/middleware/errorHandler.test.ts | 59 +++++++++++++++++++++++++---- src/middleware/errorHandler.ts | 9 +++++ src/routes/proxyRoutes.ts | 44 +++++++++++++++++++++ 3 files changed, 104 insertions(+), 8 deletions(-) diff --git a/src/middleware/errorHandler.test.ts b/src/middleware/errorHandler.test.ts index b980689..757048d 100644 --- a/src/middleware/errorHandler.test.ts +++ b/src/middleware/errorHandler.test.ts @@ -1,7 +1,7 @@ import { Request, Response, NextFunction } from 'express'; -import { errorHandler } from '../middleware/errorHandler.js'; -import { - BadRequestError, +const { errorHandler } = require('../middleware/errorHandler.js'); +import { + BadRequestError, UnauthorizedError, ForbiddenError, NotFoundError, @@ -33,12 +33,14 @@ describe('Error Handler', () => { mockRes = { status: jest.fn().mockReturnThis(), json: jest.fn(), - headersSent: false + destroy: jest.fn(), + headersSent: false, + writableEnded: false, }; mockNext = jest.fn(); }); - afterEach(() => { + afterEach((() => { jest.clearAllMocks(); }); @@ -48,7 +50,8 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, + mockRes + as Response, mockNext ); @@ -133,12 +136,52 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, + mockRes + as Response, + mockNext + ); + + expect(mockRes.status).not.toHaveBeenCalled(); + expect(mockRes.json).not.toHaveBeenCalled(); + }); + + it('destroys the socket when an error occurs after headers are sent', () => { + mockRes.headersSent = true; + mockRes.writableEnded = false; + const error = new Error('mid-stream failure'); + + errorHandler( + error, + mockReq as Request, + mockRes + as Response, mockNext ); expect(mockRes.status).not.toHaveBeenCalled(); expect(mockRes.json).not.toHaveBeenCalled(); + expect(mockRes.destroy).toHaveBeenCalledWith(error); + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + '[errorHandler]', + expect.objectContaining({ requestId: 'test-request-id' }) + ); + }); + + it('does not destroy the socket if the response is already ended', () => { + mockRes.headersSent = true; + mockRes.writableEnded = true; + const error = new Error('late error'); + + errorHandler( + error, + mockReq as Request, + mockRes as Response, + mockNext + ); + + expect(mockRes.destroy).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledTimes(1); }); it('should include explicit catalog code when provided', () => { @@ -212,7 +255,7 @@ describe('Error Handler', () => { expect(call.error.code).toBe('TOO_MANY_REQUESTS'); }); - it('all error envelopes have required fields', () => { + it('all error envelopes have required field', () => { const error = new BadRequestError('test'); errorHandler(error, mockReq as Request, mockRes as Response, mockNext); diff --git a/src/middleware/errorHandler.ts b/src/middleware/errorHandler.ts index ebe5a38..e651d6a 100644 --- a/src/middleware/errorHandler.ts +++ b/src/middleware/errorHandler.ts @@ -32,6 +32,7 @@ function extractValidationDetails(err: unknown): ValidationErrorDetail[] | undef * - Returns consistent JSON envelope: { success: false, error: { code, message }, requestId, timestamp } * - Never sends stack traces to the client in production * - Logs full error server-side + * - If headers are already sent, destroys the socket so the client sees a terminated stream */ export function errorHandler( err: unknown, @@ -84,4 +85,12 @@ export function errorHandler( } else { logger.error("[errorHandler]", logData); } + + if (res.headersSent) { + // Headers are already on the wire; we cannot write an error envelope. + // Terminate the connection so the client observes a truncated stream instead of hanging. + if (!res.writableEnded) { + res.destroy(err instance of Error ? err : undefined); + } + } } diff --git a/src/routes/proxyRoutes.ts b/src/routes/proxyRoutes.ts index 95aa201..517e19e 100644 --- a/src/routes/proxyRoutes.ts +++ b/src/routes/proxyRoutes.ts @@ -213,6 +213,7 @@ export function createProxyRouter(deps: ProxyDeps): Router { // 7. Proxy with circuit breaker and timeout let upstreamStatus = 502; const timer = startUpstreamTimer(apiEntry.id, req.method); + let headersFlushed = false; try { const executeWithRetry = async (attempt = 1): Promise => { @@ -254,6 +255,7 @@ export function createProxyRouter(deps: ProxyDeps): Router { // Stream body back res.status(upstreamStatus); + headersFlushed = true; if (upstreamRes.body) { const reader = upstreamRes.body.getReader(); const pump = async (): Promise => { @@ -268,6 +270,7 @@ export function createProxyRouter(deps: ProxyDeps): Router { } else { const text = await upstreamRes.text(); res.send(text); + headersFlushed = true; } } catch (err: unknown) { let outcome: UpstreamOutcome = 'error'; @@ -279,6 +282,14 @@ export function createProxyRouter(deps: ProxyDeps): Router { // Update metric await circuitBreaker.getMetrics(breakerKey); setGatewayUpstreamBreakerState(breakerKey, 1); + if (headersFlushed || res.headersSent) { + logger.error( + { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, + 'Proxy error after headers flushed; destroying response socket', + ); + res.destroy(err instanceof Error ? err : undefined); + return; + } throw new BadGatewayError('Bad Gateway: upstream unavailable'); } else if (err instanceof DOMException && err.name === 'TimeoutError') { upstreamStatus = 504; @@ -288,6 +299,14 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); + if (headersFlushed || res.headersSent) { + logger.error( + { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, + 'Proxy error after headers flushed; destroying response socket', + ); + res.destroy(err instanceof Error ? err : undefined); + return; + } throw new GatewayTimeoutError('Upstream service timed out'); } else if (err instanceof TypeError && (err as NodeJS.ErrnoException).code === 'UND_ERR_CONNECT_TIMEOUT') { upstreamStatus = 504; @@ -297,6 +316,14 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); + if (headersFlushed || res.headersSent) { + logger.error( + { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, + 'Proxy error after headers flushed; destroying response socket', + ); + res.destroy(err instanceof Error ? err : undefined); + return; + } throw new GatewayTimeoutError('Upstream service timed out'); } else { upstreamStatus = 502; @@ -305,6 +332,14 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); + if (headersFlushed || res.headersSent) { + logger.error( + { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, + 'Proxy error after headers flushed; destroying response socket', + ); + res.destroy(err instanceof Error ? err : undefined); + return; + } throw new BadGatewayError('Bad Gateway: upstream unreachable'); } } @@ -398,6 +433,15 @@ export function createProxyRouter(deps: ProxyDeps): Router { }); } } catch (error) { + const requestId = req.id || getOrCreateRequestId(randomUUID); + if (res.headersSent) { + logger.error( + { requestId, err: error }, + 'Proxy error after headers flushed; destroying response socket', + ); + res.destroy(error instanceof Error ? error : undefined); + return; + } next(error); } } From 2df7282d110eeebd153645ee3d54bee5d6daaa6b Mon Sep 17 00:00:00 2001 From: Ogunleye Pelumi Date: Tue, 29 Sep 2026 22:16:27 +0100 Subject: [PATCH 2/3] security: Destroy sockets when errors occur after headers flush (#1277) --- src/middleware/errorHandler.test.ts | 75 +++++------- src/middleware/errorHandler.ts | 14 +-- src/routes/proxyRoutes.ts | 175 +++++----------------------- 3 files changed, 64 insertions(+), 200 deletions(-) diff --git a/src/middleware/errorHandler.test.ts b/src/middleware/errorHandler.test.ts index 757048d..2d9e081 100644 --- a/src/middleware/errorHandler.test.ts +++ b/src/middleware/errorHandler.test.ts @@ -1,5 +1,5 @@ import { Request, Response, NextFunction } from 'express'; -const { errorHandler } = require('../middleware/errorHandler.js'); +const { errorHandler } = require('./errorHandler.js'); import { BadRequestError, UnauthorizedError, @@ -9,8 +9,9 @@ import { TooManyRequestsError, AppError, } from '../errors/index.js'; -import { ValidationError } from '../middleware/validate.js'; +const { ValidationError } = require('./validate.js'); import { logger } from '../logger.js'; +const { errorHandler } = require('./errorHandler.js'); import type { ErrorEnvelope } from '../types/ResponseEnvelope.js'; jest.mock('../logger.js', () => ({ @@ -51,13 +52,11 @@ describe('Error Handler', () => { error, mockReq as Request, mockRes - as Response, - mockNext ); expect(mockRes.status).toHaveBeenCalledWith(400); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -66,7 +65,7 @@ describe('Error Handler', () => { code: 'BAD_REQUEST', message: 'Test bad request', }); - expect(typeof call.timestamp).toBe('string'); + expect(typeof call.timestamp).toBe(String); expect(logger.error).toHaveBeenCalledWith( '[errorHandler]', @@ -80,13 +79,12 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, - mockNext + mockRes ); expect(mockRes.status).toHaveBeenCalledWith(500); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -102,13 +100,12 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, - mockNext + mockRes ); expect(mockRes.status).toHaveBeenCalledWith(500); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.success).toBe(false); expect(call.error.code).toBe('INTERNAL_SERVER_ERROR'); }); @@ -121,11 +118,10 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, - mockNext + mockRes ); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.requestId).toBe('unknown'); }); @@ -137,51 +133,43 @@ describe('Error Handler', () => { error, mockReq as Request, mockRes - as Response, - mockNext ); expect(mockRes.status).not.toHaveBeenCalled(); expect(mockRes.json).not.toHaveBeenCalled(); }); - it('destroys the socket when an error occurs after headers are sent', () => { + it('should destroy the socket when an error occurs after headers are sent', () => { mockRes.headersSent = true; - mockRes.writableEnded = false; const error = new Error('mid-stream failure'); errorHandler( error, mockReq as Request, mockRes - as Response, - mockNext ); expect(mockRes.status).not.toHaveBeenCalled(); expect(mockRes.json).not.toHaveBeenCalled(); expect(mockRes.destroy).toHaveBeenCalledWith(error); - expect(logger.error).toHaveBeenCalledTimes(1); expect(logger.error).toHaveBeenCalledWith( '[errorHandler]', - expect.objectContaining({ requestId: 'test-request-id' }) + expect.objectContaining({ requestId: 'test-request-id', statusCode: 500 }) ); }); - it('does not destroy the socket if the response is already ended', () => { + it('should not destroy the socket when the response is already ended', () => { mockRes.headersSent = true; mockRes.writableEnded = true; - const error = new Error('late error'); + const error = new Error('late failure'); errorHandler( error, mockReq as Request, - mockRes as Response, - mockNext + mockRes ); expect(mockRes.destroy).not.toHaveBeenCalled(); - expect(logger.error).toHaveBeenCalledTimes(1); }); it('should include explicit catalog code when provided', () => { @@ -190,12 +178,11 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes as Response, - mockNext + mockRes ); expect(mockRes.status).toHaveBeenCalledWith(422); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -215,51 +202,51 @@ describe('Error Handler', () => { }, ]); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); expect(mockRes.status).toHaveBeenCalledWith(400); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.error.details).toBeDefined(); expect(Array.isArray(call.error.details)).toBe(true); }); it('should map ForbiddenError to 403', () => { const error = new ForbiddenError('Test forbidden'); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); expect(mockRes.status).toHaveBeenCalledWith(403); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.error.code).toBe('FORBIDDEN'); }); it('should map NotFoundError to 404', () => { const error = new NotFoundError('Test not found'); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); expect(mockRes.status).toHaveBeenCalledWith(404); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.error.code).toBe('NOT_FOUND'); }); it('should map PaymentRequiredError to 402', () => { const error = new PaymentRequiredError('Test payment required'); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); expect(mockRes.status).toHaveBeenCalledWith(402); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.error.code).toBe('PAYMENT_REQUIRED'); }); it('should map TooManyRequestsError to 429', () => { const error = new TooManyRequestsError('Test too many requests'); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); expect(mockRes.status).toHaveBeenCalledWith(429); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call.error.code).toBe('TOO_MANY_REQUESTS'); }); - it('all error envelopes have required field', () => { + it('all error envelopes have required fields', () => { const error = new BadRequestError('test'); - errorHandler(error, mockReq as Request, mockRes as Response, mockNext); + errorHandler(error, mockReq as Request, mockRes); - const call = (mockRes.json as jest.Mock).mock.calls[0][0]; + const call = (mockRes.json as jest.Mock).mockResults[0][0]; expect(call).toHaveProperty('success'); expect(call).toHaveProperty('requestId'); expect(call).toHaveProperty('timestamp'); diff --git a/src/middleware/errorHandler.ts b/src/middleware/errorHandler.ts index e651d6a..33c2419 100644 --- a/src/middleware/errorHandler.ts +++ b/src/middleware/errorHandler.ts @@ -1,11 +1,7 @@ import type { Request, Response, NextFunction } from 'express'; -import { isAppError } from '../errors/index.js'; -import { logger } from '../logger.js'; +import { isAppError } from '../errors/index.js';import { logger } from '../logger.js'; import type { ValidationErrorDetail } from './validate.js'; -import { ValidationError } from './validate.js'; -import { buildErrorEnvelope } from './envelope.js'; -import type { ErrorEnvelope } from '../types/ResponseEnvelope.js'; -import { normalizeError } from '../errors/errorEnvelopePolicy.js'; +import { ValidationError } from './validate.js';import { buildErrorEnvelope } from './envelope.js';import type { ErrorEnvelope } from '../types/ResponseEnvelope.js';import { normalizeError } from '../errors/errorEnvelopePolicy.js'; const isProduction = process.env.NODE_ENV === "production"; @@ -32,7 +28,7 @@ function extractValidationDetails(err: unknown): ValidationErrorDetail[] | undef * - Returns consistent JSON envelope: { success: false, error: { code, message }, requestId, timestamp } * - Never sends stack traces to the client in production * - Logs full error server-side - * - If headers are already sent, destroys the socket so the client sees a terminated stream + * - When headers are already sent, destroys the socket so clients see a terminated stream */ export function errorHandler( err: unknown, @@ -87,10 +83,10 @@ export function errorHandler( } if (res.headersSent) { - // Headers are already on the wire; we cannot write an error envelope. + // Headers are already on the wire; we can't write a JSON error envelope. // Terminate the connection so the client observes a truncated stream instead of hanging. if (!res.writableEnded) { - res.destroy(err instance of Error ? err : undefined); + res.destroy(err instanceof Error ? err : undefined); } } } diff --git a/src/routes/proxyRoutes.ts b/src/routes/proxyRoutes.ts index 517e19e..cec51f2 100644 --- a/src/routes/proxyRoutes.ts +++ b/src/routes/proxyRoutes.ts @@ -1,7 +1,6 @@ import { Router, Request, Response, NextFunction } from 'express'; import { randomUUID } from 'node:crypto'; -import { ProxyDeps, ProxyConfig, ApiRegistryEntry, EndpointPricing } from '../types/gateway.js'; -import { resolveEndpointPrice } from '../data/apiRegistry.js'; +import { ProxyDeps, ProxyConfig, ApiRegistryEntry, EndpointPricing } from '../types/gateway.js';import { resolveEndpointPrice } from '../data/apiRegistry.js'; import { startUpstreamTimer, recordProxyPrematureAbort, @@ -9,23 +8,19 @@ import { setGatewayUpstreamBreakerState, recordEndpointThroughputSaturation, } from '../metrics.js'; -import { createMapBackedGatewayApiKeyAuthMiddleware } from '../middleware/gatewayApiKeyAuth.js'; -import { createConfiguredGatewayRateLimitMiddleware } from '../middleware/gatewayRateLimit.js'; -import { buildHopByHopSet } from '../lib/hopByHop.js'; -import { +import { createMapBackedGatewayApiKeyAuthMiddleware } from '../middleware/gatewayApiKeyAuth.js';import { createConfiguredGatewayRateLimitMiddleware } from '../middleware/gatewayRateLimit.js'; +import { buildHopByHopSet } from '../lib/hopByHop.js';import { buildUpstreamTargetUrl, DEFAULT_UPSTREAM_HOST_ALLOWLIST, validateResolvedUpstreamTarget, -} from '../lib/upstreamTarget.js'; -import { +} from '../lib/upstreamTarget.js';import { BadGatewayError, GatewayTimeoutError, InternalServerError, PaymentRequiredError, ServiceUnavailableError, TooManyRequestsError, -} from '../errors/index.js'; -import { CircuitBreakerOpenError } from '../lib/errors.js'; +} from '../errors/index.js';import { CircuitBreakerOpenError } from '../lib/errors.js'; import { CircuitBreaker } from '../lib/circuitBreaker.js'; import { env } from '../config/env.js'; import { getOrCreateRequestId } from '../utils/asyncContext.js'; @@ -35,7 +30,7 @@ import { logger } from '../logger.js'; /** * Headers that must never be forwarded to the upstream server. * - * Includes all RFC 7230 §6.1 hop-by-hop headers plus gateway-specific + * Includes all RFC 7230 ¦6.1 hop-by-hop headers plus gateway-specific * internal headers (host, x-api-key) that must not leak to the origin. * Dynamic Connection-listed headers are stripped at request time via * buildHopByHopSet(). @@ -117,8 +112,8 @@ export function createProxyRouter(deps: ProxyDeps): Router { }); // Per-user token-bucket rate limiter (issue #870). - // Runs AFTER authMiddleware so req.apiKeyRecord.userId is guaranteed to be - // populated. Reads limits from GATEWAY_RATE_LIMIT_* env vars by default; + // Runs AFTER authMiddleware so req.apiKeyRecord.userId is guaranteed to + // be populated. Reads limits from GATEWAY_RATE_LIMIT_* env vars by default; // can be overridden via deps for testing. const gatewayRateLimitMiddleware = deps.gatewayRateLimitMiddleware ?? createConfiguredGatewayRateLimitMiddleware(); @@ -194,7 +189,7 @@ export function createProxyRouter(deps: ProxyDeps): Router { // 6. Build forwarded headers — strip hop-by-hop and gateway-internal headers. // buildHopByHopSet() also strips any additional names listed in the - // incoming Connection header value (RFC 7230 §6.1). + // incoming Connection header value (RFC 7230 ¦6.1). const forwardHeaders: Record = {}; const connectionValue = typeof req.headers['connection'] === 'string' ? req.headers['connection'] @@ -213,7 +208,6 @@ export function createProxyRouter(deps: ProxyDeps): Router { // 7. Proxy with circuit breaker and timeout let upstreamStatus = 502; const timer = startUpstreamTimer(apiEntry.id, req.method); - let headersFlushed = false; try { const executeWithRetry = async (attempt = 1): Promise => { @@ -255,7 +249,6 @@ export function createProxyRouter(deps: ProxyDeps): Router { // Stream body back res.status(upstreamStatus); - headersFlushed = true; if (upstreamRes.body) { const reader = upstreamRes.body.getReader(); const pump = async (): Promise => { @@ -270,7 +263,6 @@ export function createProxyRouter(deps: ProxyDeps): Router { } else { const text = await upstreamRes.text(); res.send(text); - headersFlushed = true; } } catch (err: unknown) { let outcome: UpstreamOutcome = 'error'; @@ -282,14 +274,6 @@ export function createProxyRouter(deps: ProxyDeps): Router { // Update metric await circuitBreaker.getMetrics(breakerKey); setGatewayUpstreamBreakerState(breakerKey, 1); - if (headersFlushed || res.headersSent) { - logger.error( - { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, - 'Proxy error after headers flushed; destroying response socket', - ); - res.destroy(err instanceof Error ? err : undefined); - return; - } throw new BadGatewayError('Bad Gateway: upstream unavailable'); } else if (err instanceof DOMException && err.name === 'TimeoutError') { upstreamStatus = 504; @@ -299,14 +283,6 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); - if (headersFlushed || res.headersSent) { - logger.error( - { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, - 'Proxy error after headers flushed; destroying response socket', - ); - res.destroy(err instanceof Error ? err : undefined); - return; - } throw new GatewayTimeoutError('Upstream service timed out'); } else if (err instanceof TypeError && (err as NodeJS.ErrnoException).code === 'UND_ERR_CONNECT_TIMEOUT') { upstreamStatus = 504; @@ -316,133 +292,38 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); - if (headersFlushed || res.headersSent) { - logger.error( - { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, - 'Proxy error after headers flushed; destroying response socket', - ); - res.destroy(err instanceof Error ? err : undefined); - return; - } throw new GatewayTimeoutError('Upstream service timed out'); - } else { - upstreamStatus = 502; + } else if (err instanceof TypeError && (err as NodeJS.ErrnoException).code === 'UND_ERR_CONNECT_TIMEOUT') { + upstreamStatus = 504; + outcome = 'timeout'; timer.stop(upstreamStatus, outcome); // Update metric after failure const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); - if (headersFlushed || res.headersSent) { + throw new GatewayTimeoutError('Upstream service timed out'); + } else { + // Mid-stream failure after headers have been flushed: the client has + // already received a partial response, so we cannot send an error + // body. Terminate the socket instead of leaving a truncated stream + // open, and log once with the requestId for observability. + timer.stop(upstreamStatus, outcome); + if (res.headersSent) { logger.error( - { requestId, err, upstreamStatus, apiId: String(apiEntry.id) }, - 'Proxy error after headers flushed; destroying response socket', + { requestId, err }, + 'Proxy error after headers sent; destroying connection', ); res.destroy(err instanceof Error ? err : undefined); return; } - throw new BadGatewayError('Bad Gateway: upstream unreachable'); + throw err; } } - - // 8. Keep metering and billing consistent — but ONLY after the response - // has been fully delivered to the caller. - // - // We distinguish two response lifecycle events: - // • 'finish' — Node/Express has flushed all data and ended the - // response normally. This is the success path; we - // record usage here. - // • 'close' — The underlying socket was torn down. When this - // fires WITHOUT a prior 'finish' it means the client - // disconnected mid-stream (premature abort). In that - // case we must NOT record usage because the caller - // never received the response. - // - // Using a one-shot 'finish' listener (registered before we start - // streaming) ensures we capture the event even if the stream - // completes synchronously. The 'close' listener is a guard that - // cancels the deferred work when the socket drops first. - if (config.recordableStatuses(upstreamStatus)) { - // Track whether the response finished cleanly before the socket closed. - let responseFinished = false; - - res.once('finish', () => { - responseFinished = true; - - // Run usage recording in a non-blocking microtask so it does not - // delay the event loop that is already handling the next request. - setImmediate(() => { - void (async () => { - try { - const recorded = await usageStore.record({ - id: randomUUID(), // ID of the usage event itself - requestId, // Idempotency key — prevents double-counts - // apiKey field omitted to prevent storing plaintext keys - apiKeyId: keyRecord.id, - apiId: String(apiEntry.id), - endpointId: endpoint.endpointId, - userId: keyRecord.userId, - amountUsdc: endpoint.priceUsdc, - statusCode: upstreamStatus, - timestamp: new Date().toISOString(), - }); - - if (recorded) { - defaultUsageSseBroadcaster.emitForUser(keyRecord.userId, { - id: randomUUID(), - requestId, - // apiKey field omitted to prevent broadcasting plaintext keys - apiKeyId: keyRecord.id, - apiId: String(apiEntry.id), - endpointId: endpoint.endpointId, - userId: keyRecord.userId, - amountUsdc: endpoint.priceUsdc, - statusCode: upstreamStatus, - timestamp: new Date().toISOString(), - }); - } - - recordEndpointThroughputSaturation({ - apiId: String(apiEntry.id), - endpointId: endpoint.endpointId, - endpointPath: endpoint.path, - advertisedLimitPerMinute: Number(keyRecord?.rateLimitPerMinute ?? 0), - observedAt: Date.now(), - }); - - // Only deduct billing if this requestId hasn't been processed - // before (idempotency guard inside usageStore.record). - if (recorded && endpoint.priceUsdc > 0) { - billing.deductCredit(keyRecord.userId, endpoint.priceUsdc).catch((err) => { - console.error('Background billing deduction failed:', err); - }); - } - } catch (err) { - console.error('Background usage recording failed:', err); - } - })(); - }); - }); - - res.once('close', () => { - // 'close' fires after 'finish' on a normal response, or on its own - // when the socket is destroyed prematurely. Only treat it as an - // abort when 'finish' has NOT already fired. - if (!responseFinished) { - recordProxyPrematureAbort(); - } - }); - } - } catch (error) { - const requestId = req.id || getOrCreateRequestId(randomUUID); - if (res.headersSent) { - logger.error( - { requestId, err: error }, - 'Proxy error after headers flushed; destroying response socket', - ); - res.destroy(error instanceof Error ? error : undefined); - return; - } - next(error); + } catch (err: unknown) { + // Errors that reach here occurred before any body bytes were written + // (or were re-thrown from the inner catch). Delegate to the central + // error handler, which owns the logging and response formatting. + next(err); } } From 96d757b52d54f643e3b0837abf50080b6c248037 Mon Sep 17 00:00:00 2001 From: Ogunleye Pelumi Date: Tue, 29 Sep 2026 22:17:56 +0100 Subject: [PATCH 3/3] security: Destroy sockets when errors occur after headers flush (#1277) --- src/middleware/errorHandler.test.ts | 95 ++++++++-------- src/middleware/errorHandler.ts | 17 ++- src/routes/proxyRoutes.ts | 161 ++++++++++++++++++++++------ 3 files changed, 185 insertions(+), 88 deletions(-) diff --git a/src/middleware/errorHandler.test.ts b/src/middleware/errorHandler.test.ts index 2d9e081..6409e1a 100644 --- a/src/middleware/errorHandler.test.ts +++ b/src/middleware/errorHandler.test.ts @@ -1,7 +1,6 @@ -import { Request, Response, NextFunction } from 'express'; -const { errorHandler } = require('./errorHandler.js'); -import { - BadRequestError, +import { Request, Response, NextFunction } from 'express';import { errorHandler } from '../middleware/errorHandler.js'; +import { + BadRequestError, UnauthorizedError, ForbiddenError, NotFoundError, @@ -9,9 +8,7 @@ import { TooManyRequestsError, AppError, } from '../errors/index.js'; -const { ValidationError } = require('./validate.js'); -import { logger } from '../logger.js'; -const { errorHandler } = require('./errorHandler.js'); +import { ValidationError } from '../middleware/validate.js';import { logger } from '../logger.js'; import type { ErrorEnvelope } from '../types/ResponseEnvelope.js'; jest.mock('../logger.js', () => ({ @@ -24,7 +21,7 @@ jest.mock('../logger.js', () => ({ describe('Error Handler', () => { let mockReq: Partial & { id?: string }; - let mockRes: Partial; + let mockRes: Partial & { destroy?: jest.Mock }; let mockNext: NextFunction; beforeEach(() => { @@ -35,13 +32,12 @@ describe('Error Handler', () => { status: jest.fn().mockReturnThis(), json: jest.fn(), destroy: jest.fn(), - headersSent: false, - writableEnded: false, + headersSent: false }; mockNext = jest.fn(); }); - afterEach((() => { + afterEach(() => { jest.clearAllMocks(); }); @@ -51,12 +47,13 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).toHaveBeenCalledWith(400); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -65,7 +62,7 @@ describe('Error Handler', () => { code: 'BAD_REQUEST', message: 'Test bad request', }); - expect(typeof call.timestamp).toBe(String); + expect(typeof call.timestamp).toBe(typeof 'string'); expect(logger.error).toHaveBeenCalledWith( '[errorHandler]', @@ -79,12 +76,13 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).toHaveBeenCalledWith(500); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -100,12 +98,13 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).toHaveBeenCalledWith(500); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.success).toBe(false); expect(call.error.code).toBe('INTERNAL_SERVER_ERROR'); }); @@ -118,10 +117,11 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.requestId).toBe('unknown'); }); @@ -132,44 +132,46 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).not.toHaveBeenCalled(); expect(mockRes.json).not.toHaveBeenCalled(); }); - it('should destroy the socket when an error occurs after headers are sent', () => { + it('should destroy the socket when headers are already sent', () => { mockRes.headersSent = true; const error = new Error('mid-stream failure'); errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).not.toHaveBeenCalled(); expect(mockRes.json).not.toHaveBeenCalled(); expect(mockRes.destroy).toHaveBeenCalledWith(error); - expect(logger.error).toHaveBeenCalledWith( - '[errorHandler]', - expect.objectContaining({ requestId: 'test-request-id', statusCode: 500 }) - ); }); - it('should not destroy the socket when the response is already ended', () => { + it('logs the error once with requestId when headers are already sent', () => { mockRes.headersSent = true; - mockRes.writableEnded = true; - const error = new Error('late failure'); + const error = new Error('mid-stream failure'); errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); - expect(mockRes.destroy).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + '[errorHandler]', + expect.objectContaining({ requestId: 'test-request-id' }) + ); }); it('should include explicit catalog code when provided', () => { @@ -178,11 +180,12 @@ describe('Error Handler', () => { errorHandler( error, mockReq as Request, - mockRes + mockRes as Response, + mockNext ); expect(mockRes.status).toHaveBeenCalledWith(422); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call).toMatchObject({ success: false, requestId: 'test-request-id', @@ -202,51 +205,51 @@ describe('Error Handler', () => { }, ]); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); expect(mockRes.status).toHaveBeenCalledWith(400); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.error.details).toBeDefined(); expect(Array.isArray(call.error.details)).toBe(true); }); it('should map ForbiddenError to 403', () => { const error = new ForbiddenError('Test forbidden'); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); expect(mockRes.status).toHaveBeenCalledWith(403); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.error.code).toBe('FORBIDDEN'); }); it('should map NotFoundError to 404', () => { const error = new NotFoundError('Test not found'); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); expect(mockRes.status).toHaveBeenCalledWith(404); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.error.code).toBe('NOT_FOUND'); }); it('should map PaymentRequiredError to 402', () => { const error = new PaymentRequiredError('Test payment required'); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); expect(mockRes.status).toHaveBeenCalledWith(402); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.error.code).toBe('PAYMENT_REQUIRED'); }); it('should map TooManyRequestsError to 429', () => { const error = new TooManyRequestsError('Test too many requests'); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); expect(mockRes.status).toHaveBeenCalledWith(429); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call.error.code).toBe('TOO_MANY_REQUESTS'); }); it('all error envelopes have required fields', () => { const error = new BadRequestError('test'); - errorHandler(error, mockReq as Request, mockRes); + errorHandler(error, mockReq as Request, mockRes as Response, mockNext); - const call = (mockRes.json as jest.Mock).mockResults[0][0]; + const call = (mockRes.json as jest.Mock).mock.calls[0][0]; expect(call).toHaveProperty('success'); expect(call).toHaveProperty('requestId'); expect(call).toHaveProperty('timestamp'); diff --git a/src/middleware/errorHandler.ts b/src/middleware/errorHandler.ts index 33c2419..b1a4130 100644 --- a/src/middleware/errorHandler.ts +++ b/src/middleware/errorHandler.ts @@ -28,7 +28,7 @@ function extractValidationDetails(err: unknown): ValidationErrorDetail[] | undef * - Returns consistent JSON envelope: { success: false, error: { code, message }, requestId, timestamp } * - Never sends stack traces to the client in production * - Logs full error server-side - * - When headers are already sent, destroys the socket so clients see a terminated stream + * - When headers are already sent, destroys the socket so the client sees a terminated stream */ export function errorHandler( err: unknown, @@ -63,6 +63,13 @@ export function errorHandler( if (!res.headersSent) { res.status(statusCode).json(body); + } else { + // Headers already flushed: we cannot write a JSON envelope. + // Terminate the socket so the client observes a truncated stream + // instead of hanging until its own timeout. + if (typeof res.destroy === 'function') { + res.destroy(err instanceof Error ? err : undefined); + } } const logData = { @@ -81,12 +88,4 @@ export function errorHandler( } else { logger.error("[errorHandler]", logData); } - - if (res.headersSent) { - // Headers are already on the wire; we can't write a JSON error envelope. - // Terminate the connection so the client observes a truncated stream instead of hanging. - if (!res.writableEnded) { - res.destroy(err instanceof Error ? err : undefined); - } - } } diff --git a/src/routes/proxyRoutes.ts b/src/routes/proxyRoutes.ts index cec51f2..8ea2627 100644 --- a/src/routes/proxyRoutes.ts +++ b/src/routes/proxyRoutes.ts @@ -1,6 +1,7 @@ import { Router, Request, Response, NextFunction } from 'express'; import { randomUUID } from 'node:crypto'; -import { ProxyDeps, ProxyConfig, ApiRegistryEntry, EndpointPricing } from '../types/gateway.js';import { resolveEndpointPrice } from '../data/apiRegistry.js'; +import { ProxyDeps, ProxyConfig, ApiRegistryEntry, EndpointPricing } from '../types/gateway.js'; +import { resolveEndpointPrice } from '../data/apiRegistry.js'; import { startUpstreamTimer, recordProxyPrematureAbort, @@ -8,19 +9,23 @@ import { setGatewayUpstreamBreakerState, recordEndpointThroughputSaturation, } from '../metrics.js'; -import { createMapBackedGatewayApiKeyAuthMiddleware } from '../middleware/gatewayApiKeyAuth.js';import { createConfiguredGatewayRateLimitMiddleware } from '../middleware/gatewayRateLimit.js'; -import { buildHopByHopSet } from '../lib/hopByHop.js';import { +import { createMapBackedGatewayApiKeyAuthMiddleware } from '../middleware/gatewayApiKeyAuth.js'; +import { createConfiguredGatewayRateLimitMiddleware } from '../middleware/gatewayRateLimit.js'; +import { buildHopByHopSet } from '../lib/hopByHop.js'; +import { buildUpstreamTargetUrl, DEFAULT_UPSTREAM_HOST_ALLOWLIST, validateResolvedUpstreamTarget, -} from '../lib/upstreamTarget.js';import { +} from '../lib/upstreamTarget.js'; +import { BadGatewayError, GatewayTimeoutError, InternalServerError, PaymentRequiredError, ServiceUnavailableError, TooManyRequestsError, -} from '../errors/index.js';import { CircuitBreakerOpenError } from '../lib/errors.js'; +} from '../errors/index.js'; +import { CircuitBreakerOpenError } from '../lib/errors.js'; import { CircuitBreaker } from '../lib/circuitBreaker.js'; import { env } from '../config/env.js'; import { getOrCreateRequestId } from '../utils/asyncContext.js'; @@ -30,7 +35,7 @@ import { logger } from '../logger.js'; /** * Headers that must never be forwarded to the upstream server. * - * Includes all RFC 7230 ¦6.1 hop-by-hop headers plus gateway-specific + * Includes all RFC 7230 §6.1 hop-by-hop headers plus gateway-specific * internal headers (host, x-api-key) that must not leak to the origin. * Dynamic Connection-listed headers are stripped at request time via * buildHopByHopSet(). @@ -112,8 +117,8 @@ export function createProxyRouter(deps: ProxyDeps): Router { }); // Per-user token-bucket rate limiter (issue #870). - // Runs AFTER authMiddleware so req.apiKeyRecord.userId is guaranteed to - // be populated. Reads limits from GATEWAY_RATE_LIMIT_* env vars by default; + // Runs AFTER authMiddleware so req.apiKeyRecord.userId is guaranteed to be + // populated. Reads limits from GATEWAY_RATE_LIMIT_* env vars by default; // can be overridden via deps for testing. const gatewayRateLimitMiddleware = deps.gatewayRateLimitMiddleware ?? createConfiguredGatewayRateLimitMiddleware(); @@ -189,7 +194,7 @@ export function createProxyRouter(deps: ProxyDeps): Router { // 6. Build forwarded headers — strip hop-by-hop and gateway-internal headers. // buildHopByHopSet() also strips any additional names listed in the - // incoming Connection header value (RFC 7230 ¦6.1). + // incoming Connection header value (RFC 7230 §6.1). const forwardHeaders: Record = {}; const connectionValue = typeof req.headers['connection'] === 'string' ? req.headers['connection'] @@ -267,6 +272,26 @@ export function createProxyRouter(deps: ProxyDeps): Router { } catch (err: unknown) { let outcome: UpstreamOutcome = 'error'; + // If headers have already been flushed to the client, we cannot send a + // structured error response. Destroy the socket so the client sees a + // terminated stream instead of hanging on a truncated body. Log once + // with the requestId for observability. + if (res.headersSent) { + logger.error( + { + err, + requestId, + apiId: String(apiEntry.id), + endpointId: endpoint.endpointId, + upstreamStatus, + }, + 'Proxy error after headers sent; destroying response socket', + ); + timer.stop(upstreamStatus, outcome); + res.destroy(err instanceof Error ? err : undefined); + return; + } + if (err instanceof CircuitBreakerOpenError) { // Circuit breaker open — don't bill the caller upstreamStatus = 502; @@ -293,37 +318,107 @@ export function createProxyRouter(deps: ProxyDeps): Router { const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); throw new GatewayTimeoutError('Upstream service timed out'); - } else if (err instanceof TypeError && (err as NodeJS.ErrnoException).code === 'UND_ERR_CONNECT_TIMEOUT') { - upstreamStatus = 504; - outcome = 'timeout'; + } else { + upstreamStatus = 502; timer.stop(upstreamStatus, outcome); // Update metric after failure const failedMetrics = await circuitBreaker.getMetrics(breakerKey); const failedStateValue = failedMetrics.state === 'CLOSED' ? 0 : failedMetrics.state === 'OPEN' ? 1 : 2; setGatewayUpstreamBreakerState(breakerKey, failedStateValue); - throw new GatewayTimeoutError('Upstream service timed out'); - } else { - // Mid-stream failure after headers have been flushed: the client has - // already received a partial response, so we cannot send an error - // body. Terminate the socket instead of leaving a truncated stream - // open, and log once with the requestId for observability. - timer.stop(upstreamStatus, outcome); - if (res.headersSent) { - logger.error( - { requestId, err }, - 'Proxy error after headers sent; destroying connection', - ); - res.destroy(err instanceof Error ? err : undefined); - return; - } - throw err; + throw new BadGatewayError('Bad Gateway: upstream unreachable'); } } - } catch (err: unknown) { - // Errors that reach here occurred before any body bytes were written - // (or were re-thrown from the inner catch). Delegate to the central - // error handler, which owns the logging and response formatting. - next(err); + + // 8. Keep metering and billing consistent — but ONLY after the response + // has been fully delivered to the caller. + // + // We distinguish two response lifecycle events: + // • 'finish' — Node/Express has flushed all data and ended the + // response normally. This is the success path; we + // record usage here. + // • 'close' — The underlying socket was torn down. When this + // fires WITHOUT a prior 'finish' it means the client + // disconnected mid-stream (premature abort). In that + // case we must NOT record usage because the caller + // never received the response. + // + // Using a one-shot 'finish' listener (registered before we start + // streaming) ensures we capture the event even if the stream + // completes synchronously. The 'close' listener is a guard that + // cancels the deferred work when the socket drops first. + if (config.recordableStatuses(upstreamStatus)) { + // Track whether the response finished cleanly before the socket closed. + let responseFinished = false; + + res.once('finish', () => { + responseFinished = true; + + // Run usage recording in a non-blocking microtask so it does not + // delay the event loop that is already handling the next request. + setImmediate(() => { + void (async () => { + try { + const recorded = await usageStore.record({ + id: randomUUID(), // ID of the usage event itself + requestId, // Idempotency key — prevents double-counts + // apiKey field omitted to prevent storing plaintext keys + apiKeyId: keyRecord.id, + apiId: String(apiEntry.id), + endpointId: endpoint.endpointId, + userId: keyRecord.userId, + amountUsdc: endpoint.priceUsdc, + statusCode: upstreamStatus, + timestamp: new Date().toISOString(), + }); + + if (recorded) { + defaultUsageSseBroadcaster.emitForUser(keyRecord.userId, { + id: randomUUID(), + requestId, + // apiKey field omitted to prevent broadcasting plaintext keys + apiKeyId: keyRecord.id, + apiId: String(apiEntry.id), + endpointId: endpoint.endpointId, + userId: keyRecord.userId, + amountUsdc: endpoint.priceUsdc, + statusCode: upstreamStatus, + timestamp: new Date().toISOString(), + }); + } + + recordEndpointThroughputSaturation({ + apiId: String(apiEntry.id), + endpointId: endpoint.endpointId, + endpointPath: endpoint.path, + advertisedLimitPerMinute: Number(keyRecord?.rateLimitPerMinute ?? 0), + observedAt: Date.now(), + }); + + // Only deduct billing if this requestId hasn't been processed + // before (idempotency guard inside usageStore.record). + if (recorded && endpoint.priceUsdc > 0) { + billing.deductCredit(keyRecord.userId, endpoint.priceUsdc).catch((err) => { + console.error('Background billing deduction failed:', err); + }); + } + } catch (err) { + console.error('Background usage recording failed:', err); + } + })(); + }); + }); + + res.once('close', () => { + // 'close' fires after 'finish' on a normal response, or on its own + // when the socket is destroyed prematurely. Only treat it as an + // abort when 'finish' has NOT already fired. + if (!responseFinished) { + recordProxyPrematureAbort(); + } + }); + } + } catch (error) { + next(error); } }