From fd387cdb7c65565e12e3d67007dfe9ea47f1b397 Mon Sep 17 00:00:00 2001 From: isaacs Date: Tue, 20 Oct 2026 15:02:38 -0700 Subject: [PATCH] ref(deno): Use a shared request handler wrapper for `Deno.serve` Deno.serve, Bun.serve and Cloudflare each wrap a web `Request` handler with their own copy of the same steps: request data and body, the `http.server` span, streamed responses, errors and sessions. The copies drift, so a fix in one runtime does not reach the others. This adds `wrapServerRequestHandler` to `@sentry/server-utils` and moves `Deno.serve` onto it. Bun.serve and Cloudflare follow in separate PRs, so each runtime's changes can be reviewed alone. The wrapper has only what Deno needs today, and the next PRs extend it. It is exported from the shared entry, so the Node-free `@sentry/server-utils/no-diagnostic-channels` entry can use it too. The rules follow core's `node:http` server code. `OPTIONS` and `HEAD` requests get no span and no body capture, but still get request data, a session and error capture. The static asset filter applies only to `GET`. For streamed responses, the wrapper uses the stream monitor from the Bun SDK, which reads only when the client pulls and reports errors from the stream source. It replaces Deno's own copy, which read ahead of the client and ignored source errors. The existing Deno tests pass with no change. Visible changes for Deno: - An error from the source of a streamed response is now captured, and the span gets an error status. - The SDK no longer reads a streamed response ahead of the client. - Errors in `OPTIONS` and `HEAD` requests now carry the request data. - `client.address` is no longer set to the path of a Unix socket. Part of #25577 Part of JS-3975 Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 8 + packages/deno/src/integrations/deno-serve.ts | 42 ++- packages/deno/src/utils/streaming.ts | 72 ----- .../deno/src/wrap-deno-request-handler.ts | 229 ++------------- packages/deno/test/deno-serve.test.ts | 40 +++ packages/deno/test/streaming.test.ts | 127 -------- packages/server-utils/src/exports.ts | 2 + .../server-utils/src/utils/monitorStream.ts | 40 +++ .../src/utils/wrapServerRequestHandler.ts | 270 ++++++++++++++++++ .../test/utils/monitorStream.test.ts | 101 +++++++ .../utils/wrapServerRequestHandler.test.ts | 157 ++++++++++ 11 files changed, 667 insertions(+), 421 deletions(-) delete mode 100644 packages/deno/src/utils/streaming.ts delete mode 100644 packages/deno/test/streaming.test.ts create mode 100644 packages/server-utils/src/utils/monitorStream.ts create mode 100644 packages/server-utils/src/utils/wrapServerRequestHandler.ts create mode 100644 packages/server-utils/test/utils/monitorStream.test.ts create mode 100644 packages/server-utils/test/utils/wrapServerRequestHandler.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3803a5d99c30..12dd671c4c50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -98,6 +98,14 @@ Work in this release was contributed by @adityareddy-dev, @jpdefo, @Jxxunnn, @ph When an instrumented package loaded after `init()` (for example through a dynamic `import()`), the SDK added its integration even if the user had removed it. This applied to an `integrations` function that filtered it out, `defaultIntegrations: false`, and a custom `defaultIntegrations` list. The SDK now applies the same rules to a late-loaded integration as to the defaults at `init()`. To do this, it calls an `integrations` function again, once for each package that loads after `init()`. +- **ref(deno): Use the shared request handler wrapper for `Deno.serve`** + + `Deno.serve` now uses a request handler wrapper that `Bun.serve` and Cloudflare will also use. Three small changes for Deno users: + + - When the source of a streamed response fails, the SDK now captures the error and marks the span as failed. + - The SDK no longer reads a streamed response ahead of the client. + - Errors in `OPTIONS` and `HEAD` requests now carry the request data. + ## 11.8.0 ### Important Changes diff --git a/packages/deno/src/integrations/deno-serve.ts b/packages/deno/src/integrations/deno-serve.ts index 8c8a4038e5b5..425040e4607e 100644 --- a/packages/deno/src/integrations/deno-serve.ts +++ b/packages/deno/src/integrations/deno-serve.ts @@ -1,6 +1,6 @@ -import type { Event, IntegrationFn, MaxRequestBodySize, Span } from '@sentry/core'; +import type { Client, Event, Integration, IntegrationFn, MaxRequestBodySize, Span } from '@sentry/core'; import { debug, defineIntegration } from '@sentry/core'; -import type { AssertTrue, CheckServerRequestOptions } from '@sentry/core/server'; +import type { AssertTrue, CheckServerRequestOptions, ServerRequestOptions } from '@sentry/core/server'; import { DEFAULT_IGNORE_STATUS_CODES, processHttpServerTransactionEvent } from '@sentry/core/server'; import type { RequestHandlerWrapperOptions } from '../wrap-deno-request-handler'; import { wrapDenoRequestHandler } from '../wrap-deno-request-handler'; @@ -140,6 +140,7 @@ const applyHandlerWrap = request, info, serveOptions, + getRequestOptions: getDenoServeRequestOptions, } as RequestHandlerWrapperOptions, () => handler(request, info as Deno.ServeHandlerInfo), ); @@ -159,21 +160,38 @@ const instrumentedDenoServe = (serve: typeof Deno.serve): typeof Deno.serve => }, }); -const _denoServeIntegration = ((options: DenoServeIntegrationOptions = {}) => { +// Every shared option is a required key, so a new one fails to compile below until the integration +// passes it on. +type DenoServeRequestOptions = { + [K in keyof Required>]: ServerRequestOptions[K]; +}; + +interface DenoServeIntegration extends Integration { + requestOptions: DenoServeRequestOptions; +} + +/** The incoming request options of the client's `denoServeIntegration`, if it has one. */ +function getDenoServeRequestOptions(client: Client): ServerRequestOptions | undefined { + return client.getIntegrationByName(INTEGRATION_NAME)?.requestOptions; +} + +const _denoServeIntegration = ((options: DenoServeIntegrationOptions = {}): DenoServeIntegration => { // oxlint-disable-next-line typescript/no-deprecated const ignoreStatusCodes = options.ignoreStatusCodes ?? DEFAULT_IGNORE_STATUS_CODES; return { name: INTEGRATION_NAME, - maxRequestBodySize: options.maxRequestBodySize, - ignoreRequestBody: options.ignoreRequestBody, - ignoreIncomingRequests: options.ignoreIncomingRequests, - ignoreStaticAssets: options.ignoreStaticAssets ?? true, - spans: options.spans, - onSpanCreated: options.onSpanCreated, - onSpanEnd: options.onSpanEnd, - sessions: options.sessions ?? true, - sessionFlushingDelayMS: options.sessionFlushingDelayMS, + requestOptions: { + maxRequestBodySize: options.maxRequestBodySize, + ignoreRequestBody: options.ignoreRequestBody, + ignoreIncomingRequests: options.ignoreIncomingRequests, + ignoreStaticAssets: options.ignoreStaticAssets ?? true, + spans: options.spans, + onSpanCreated: options.onSpanCreated, + onSpanEnd: options.onSpanEnd, + sessions: options.sessions ?? true, + sessionFlushingDelayMS: options.sessionFlushingDelayMS, + }, processEvent(event: Event): Event | null { // Gated on this integration's own span origin so it does not filter `node:http` // transactions, which `denoHttpIntegration` owns via its own `ignoreStatusCodes`. diff --git a/packages/deno/src/utils/streaming.ts b/packages/deno/src/utils/streaming.ts deleted file mode 100644 index 0b82c16e4a47..000000000000 --- a/packages/deno/src/utils/streaming.ts +++ /dev/null @@ -1,72 +0,0 @@ -import type { Span } from '@sentry/core'; -import { classifyResponseStreaming } from '@sentry/core/server'; - -/** - * Tee a stream, and end the provided span (if any) when the stream ends. - * Returns the other side of the tee, which can be used to send the - * response to a client. - * - * `onEnd` runs right before the span ends. - */ -export async function streamResponse(span: Span | undefined, res: Response, onEnd?: () => void): Promise { - const classification = classifyResponseStreaming(res); - const endSpan = (): void => { - onEnd?.(); - span?.end(); - }; - - // not streaming, just end the span and return the response - if (!classification.isStreaming || !res.body) { - endSpan(); - return res; - } - - // Streaming response detected - monitor consumption to keep span alive - try { - return new Response(monitorStream(res.body, endSpan), { - status: res.status, - statusText: res.statusText, - headers: res.headers, - }); - } catch { - // tee() failed - handle without streaming - endSpan(); - return res; - } -} - -/** - * zero-copy monitoring of stream progress. - */ -function monitorStream( - stream: ReadableStream>, - onDone: () => void, -): ReadableStream> { - const reader = stream.getReader(); - reader.closed.then( - () => onDone(), - () => onDone(), - ); - return new ReadableStream({ - async start(controller) { - let result: ReadableStreamReadResult>; - do { - result = await reader.read(); - if (result.value) { - try { - controller.enqueue(result.value); - } catch (er) { - controller.error(er); - reader.releaseLock(); - return; - } - } - } while (!result.done); - controller.close(); - reader.releaseLock(); - }, - cancel(reason) { - return reader.cancel(reason); - }, - }); -} diff --git a/packages/deno/src/wrap-deno-request-handler.ts b/packages/deno/src/wrap-deno-request-handler.ts index ed8d6ae2b77e..4d24405d58e4 100644 --- a/packages/deno/src/wrap-deno-request-handler.ts +++ b/packages/deno/src/wrap-deno-request-handler.ts @@ -1,85 +1,14 @@ -import { - CLIENT_ADDRESS, - CLIENT_PORT, - NETWORK_PROTOCOL_NAME, - SENTRY_OP, - SENTRY_SEGMENT_NAME_SOURCE, -} from '@sentry/conventions/attributes'; -import { HTTP_SERVER } from '@sentry/conventions/op'; -import type { Client, Integration } from '@sentry/core'; -import { - captureBodyFromWinterCGRequest, - captureException, - continueTrace, - debug, - getClient, - getHttpSpanDetailsFromUrlObject, - hasSpansEnabled, - hasSpanStreamingEnabled, - httpHeadersToSpanAttributes, - HTTP_SPAN_NAME_FALLBACK, - parseStringToURLObject, - setHttpStatus, - startSpanManual, - winterCGHeadersToDict, - winterCGRequestToRequestData, - withIsolationScope, -} from '@sentry/core'; -import { getClientIPAddress, isStaticAssetRequest, startRequestSession } from '@sentry/core/server'; -import type { DenoServeIntegrationOptions } from './integrations/deno-serve'; -import { streamResponse } from './utils/streaming'; +import type { Client } from '@sentry/core'; +import { debug, getClient, withIsolationScope } from '@sentry/core'; +import type { ServerRequestOptions } from '@sentry/core/server'; +import { wrapServerRequestHandler } from '@sentry/server-utils'; export type RequestHandlerWrapperOptions = { request: Request; info: Deno.ServeHandlerInfo; serveOptions?: Deno.ServeOptions; -}; - -function captureServeError(e: unknown): void { - captureException(e, { - mechanism: { - handled: false, - type: 'auto.http.deno', - data: { function: 'serve' }, - }, - }); -} - -// A user callback that throws must not fail the request or leave the span open. -function callUserCallback(name: string, callback: () => T, fallback: T): T { - try { - return callback(); - } catch (e) { - debug.error(`Error in the Deno.serve ${name} callback.`, e); - return fallback; - } -} - -function shouldCreateSpan( - client: Client, - integration: DenoServeIntegrationOptions | undefined, - request: Request, - urlPath: string, -): boolean { - if (!(integration?.spans ?? hasSpansEnabled(client.getOptions()))) { - return false; - } - if (integration?.ignoreStaticAssets && request.method === 'GET' && isStaticAssetRequest(urlPath)) { - return false; - } - const ignoreIncomingRequests = integration?.ignoreIncomingRequests; - return !( - ignoreIncomingRequests && - callUserCallback('ignoreIncomingRequests', () => ignoreIncomingRequests(urlPath, request), false) - ); -} - -const assignIfSet = , K extends keyof T>( - obj: T, - key: K, - value: T[K] | undefined | null, -): void => { - if (value !== undefined && value !== null) obj[key] = value; + /** Reads the incoming request options from the client's `denoServeIntegration`. */ + getRequestOptions: (client: Client) => ServerRequestOptions | undefined; }; export const wrapDenoRequestHandler = ( @@ -87,7 +16,7 @@ export const wrapDenoRequestHandler = ( handler: () => Promise | Response, ): Response | Promise => { return withIsolationScope(async isolationScope => { - const { request, info } = wrapperOptions; + const { request, info, getRequestOptions } = wrapperOptions; const client = getClient(); if (!client) { @@ -100,140 +29,20 @@ export const wrapDenoRequestHandler = ( } isolationScope.setClient(client); - const integration = client.getIntegrationByName('DenoServe'); - const endSession = integration?.sessions - ? startRequestSession(client, { - requestIsolationScope: isolationScope, - sessionFlushingDelayMS: integration.sessionFlushingDelayMS, - }) - : undefined; - - if (request.method === 'OPTIONS' || request.method === 'HEAD') { - try { - return await handler(); - } catch (e) { - captureServeError(e); - throw e; - } finally { - endSession?.(); - } - } - - const urlObject = parseStringToURLObject(request.url); - const [rawName, attributes] = getHttpSpanDetailsFromUrlObject( - urlObject, - 'server', - 'auto.http.deno', - request, - undefined, - client, - ); - // With span streaming, span names have to be low cardinality, so we can't fall back to the URL. - // A `route` source means the name already is (e.g. the `/` path), so it is kept as-is. - const name = - attributes[SENTRY_SEGMENT_NAME_SOURCE] === 'route' || !hasSpanStreamingEnabled(client) - ? rawName - : request.method?.toUpperCase() || HTTP_SPAN_NAME_FALLBACK; - - const contentLength = request.headers.get('content-length'); - assignIfSet(attributes, 'http.request.body.size', contentLength && parseInt(contentLength, 10)); - assignIfSet(attributes, 'user_agent.original', request.headers.get('user-agent')); - - const headers = winterCGHeadersToDict(request.headers); - const socketIp = (info?.remoteAddr as Deno.NetAddr)?.hostname; - const dataCollection = client.getDataCollectionOptions(); - if (dataCollection.userInfo) { - // `client.address` is the originating client, so a forwarding header wins over the socket, which - // behind a proxy holds the proxy's address. The socket port is the proxy's too, so `client.port` - // stays unset then. - const forwardedAddress = getClientIPAddress(headers); - assignIfSet( - attributes, - CLIENT_ADDRESS, - forwardedAddress || socketIp || (info?.remoteAddr as Deno.UnixAddr)?.path, - ); - if (!forwardedAddress) { - assignIfSet(attributes, CLIENT_PORT, (info?.remoteAddr as Deno.NetAddr)?.port); - } - } - - // describes the OSI application-layer protocol (http), not the scheme (might be https) - attributes[NETWORK_PROTOCOL_NAME] = 'http'; - - Object.assign(attributes, httpHeadersToSpanAttributes(headers, dataCollection)); - attributes[SENTRY_OP] = HTTP_SERVER; - isolationScope.setSDKProcessingMetadata({ - normalizedRequest: winterCGRequestToRequestData(request), - ipAddress: socketIp, - }); - - const effectiveBodySize = - integration?.maxRequestBodySize ?? (dataCollection.httpBodies.includes('incomingRequest') ? 'medium' : 'none'); - const ignoreRequestBody = integration?.ignoreRequestBody; - if ( - request.method !== 'GET' && - effectiveBodySize !== 'none' && - // If the filter throws, skip the body so it cannot leak data the filter was meant to hide. - !(ignoreRequestBody && callUserCallback('ignoreRequestBody', () => ignoreRequestBody(request.url, request), true)) - ) { - await captureBodyFromWinterCGRequest(request, isolationScope, effectiveBodySize); - } - - const urlPath = urlObject ? `${urlObject.pathname}${urlObject.search}` : '/'; - const createSpan = shouldCreateSpan(client, integration, request, urlPath); - - return continueTrace( + return wrapServerRequestHandler( { - sentryTrace: request.headers.get('sentry-trace') || '', - baggage: request.headers.get('baggage'), - }, - async () => { - if (!createSpan) { - let res: Response; - try { - res = await handler(); - isolationScope.setContext('response', { - status_code: res.status, - }); - } catch (e) { - captureServeError(e); - endSession?.(); - throw e; - } - return endSession ? streamResponse(undefined, res, endSession) : res; - } - - return startSpanManual({ name, attributes }, async span => { - const { onSpanCreated, onSpanEnd } = integration ?? {}; - if (onSpanCreated) { - callUserCallback('onSpanCreated', () => onSpanCreated(span, request), undefined); - } - let res: Response; - - try { - res = await handler(); - setHttpStatus(span, res.status); - isolationScope.setContext('response', { - status_code: res.status, - }); - span.setAttributes( - httpHeadersToSpanAttributes(Object.fromEntries(res.headers), dataCollection, 'response'), - ); - } catch (e) { - span.end(); - captureServeError(e); - endSession?.(); - throw e; - } - - return streamResponse(span, res, () => { - if (onSpanEnd) { - callUserCallback('onSpanEnd', () => onSpanEnd(span, request, res), undefined); - } - endSession?.(); - }); - }); + request, + client, + isolationScope, + options: getRequestOptions(client), + origin: 'auto.http.deno', + mechanism: { handled: false, type: 'auto.http.deno', data: { function: 'serve' } }, + getSocketAddress: () => { + const remoteAddr = info?.remoteAddr as Deno.NetAddr | undefined; + return remoteAddr?.hostname ? { address: remoteAddr.hostname, port: remoteAddr.port } : undefined; + }, }, + handler, ); }); }; diff --git a/packages/deno/test/deno-serve.test.ts b/packages/deno/test/deno-serve.test.ts index 90719660ea43..e7749f3c39fb 100644 --- a/packages/deno/test/deno-serve.test.ts +++ b/packages/deno/test/deno-serve.test.ts @@ -866,6 +866,46 @@ Deno.test('Deno.serve should handle streaming responses', async () => { assertEquals(transactionEvents[0]?.contexts?.trace?.op, 'http.server'); }); +Deno.test('Deno.serve should capture an error from a streamed response source', async () => { + resetGlobals(); + const errorEvents: ErrorEvent[] = []; + const transactionEvents: TransactionEvent[] = []; + + const client = init({ + dsn: 'https://username@domain/123', + tracesSampleRate: 1, + traceLifecycle: 'static', + beforeSend: (event: ErrorEvent) => { + errorEvents.push(event); + return null; + }, + beforeSendTransaction: (event: TransactionEvent) => { + transactionEvents.push(event); + return null; + }, + }) as DenoClient; + + const server = await serveForTest(() => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('chunk1\n')); + setTimeout(() => controller.error(new Error('stream interrupted')), 10); + }, + }); + return new Response(stream, { headers: { 'content-type': 'text/event-stream' } }); + }); + + const response = await fetch(`${server.url}/stream`); + await response.text().catch(() => undefined); + await server.close(); + await client.flush(); + + assertEquals(errorEvents.length, 1); + assertEquals(errorEvents[0]?.exception?.values?.[0]?.value, 'stream interrupted'); + assertEquals(errorEvents[0]?.exception?.values?.[0]?.mechanism?.type, 'auto.http.deno'); + assertEquals(transactionEvents[0]?.contexts?.trace?.status, 'internal_error'); +}); + Deno.test('Deno.serve should work when manually capturing exceptions within handler', async () => { resetGlobals(); const errorEvents: ErrorEvent[] = []; diff --git a/packages/deno/test/streaming.test.ts b/packages/deno/test/streaming.test.ts deleted file mode 100644 index 05951459e786..000000000000 --- a/packages/deno/test/streaming.test.ts +++ /dev/null @@ -1,127 +0,0 @@ -// - -import { assertEquals, assertStrictEquals } from 'https://deno.land/std@0.212.0/assert/mod.ts'; - -import { streamResponse } from '../src/utils/streaming.ts'; - -Deno.test('cancels the source when the response reader is cancelled', async () => { - let sourceCancelReason: unknown; - - const source = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('first event')); - }, - cancel(reason) { - sourceCancelReason = reason; - }, - }); - - const span = { end() {} } as Parameters[0]; - const response = await streamResponse( - span, - new Response(source, { headers: { 'content-type': 'text/event-stream' } }), - ); - const reader = response.body!.getReader(); - const reason = new Error('client disconnected'); - - await reader.read(); - await reader.cancel(reason); - - assertStrictEquals(sourceCancelReason, reason); -}); - -Deno.test('reader.closed.then(f, f) suppresses rejection when releaseLock is called on an open stream', async () => { - // Reproduces the bug from GitHub issue #20177: - // In monitorStream, reader.releaseLock() is called while the source stream - // is still open (e.g. the error path when controller.enqueue() throws). - // Per WHATWG Streams spec, this rejects reader.closed with a TypeError. - // Using .then(onDone, onDone) handles both cases; .finally() would propagate - // the rejection as unhandled. - - let onDoneCalled = false; - - const stream = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('data')); - // intentionally not closing — stream stays open - }, - }); - - const reader = stream.getReader(); - - // This is the exact pattern from monitorStream (line 84 in streaming.ts). - // With .finally(() => onDone()), this would propagate the rejection. - reader.closed.then( - () => { - onDoneCalled = true; - }, - () => { - onDoneCalled = true; - }, - ); - - await reader.read(); - - // This is what monitorStream does on the error path (line 98) when - // controller.enqueue() throws — releaseLock while the source is still open. - reader.releaseLock(); - - let unhandledRejection: PromiseRejectionEvent | undefined; - const handler = (e: PromiseRejectionEvent): void => { - e.preventDefault(); - unhandledRejection = e; - }; - globalThis.addEventListener('unhandledrejection', handler); - - try { - await new Promise(resolve => setTimeout(resolve, 50)); - - assertEquals(onDoneCalled, true, 'onDone should have been called via the rejection handler'); - assertEquals(unhandledRejection, undefined, 'should not have caused an unhandled promise rejection'); - } finally { - globalThis.removeEventListener('unhandledrejection', handler); - } -}); - -Deno.test('calls onEnd before ending the span for a non-streamed response', async () => { - const calls: string[] = []; - const span = { end: () => calls.push('end') } as unknown as Parameters[0]; - - await streamResponse(span, new Response('{}', { headers: { 'content-type': 'application/json' } }), () => - calls.push('onEnd'), - ); - - assertEquals(calls, ['onEnd', 'end']); -}); - -Deno.test('calls onEnd before ending the span when a streamed response finishes', async () => { - const calls: string[] = []; - let resolveEnded: () => void; - const ended = new Promise(resolve => (resolveEnded = resolve)); - const span = { - end: () => { - calls.push('end'); - resolveEnded(); - }, - } as unknown as Parameters[0]; - - let sourceController: ReadableStreamDefaultController | undefined; - const source = new ReadableStream({ - start(controller) { - sourceController = controller; - controller.enqueue(new TextEncoder().encode('chunk')); - }, - }); - const response = await streamResponse( - span, - new Response(source, { headers: { 'content-type': 'text/event-stream' } }), - () => calls.push('onEnd'), - ); - - await new Promise(resolve => setTimeout(resolve, 0)); - assertEquals(calls, []); - sourceController!.close(); - assertEquals(await response.text(), 'chunk'); - await ended; - assertEquals(calls, ['onEnd', 'end']); -}); diff --git a/packages/server-utils/src/exports.ts b/packages/server-utils/src/exports.ts index 15381f95dc9b..799ed7c69d18 100644 --- a/packages/server-utils/src/exports.ts +++ b/packages/server-utils/src/exports.ts @@ -1,5 +1,7 @@ // Shared exports not using diagnostics channels export { setHttpServerSpanRouteAttribute } from './utils/setHttpServerSpanRouteAttribute'; +export { wrapServerRequestHandler } from './utils/wrapServerRequestHandler'; +export type { ServerRequestHandlerConfig } from './utils/wrapServerRequestHandler'; export { setAsyncLocalStorageAsyncContextStrategy } from './async-context'; export { openTelemetryIntegration, getOtlpTracesEndpoint } from './opentelemetry'; export * from './ai'; diff --git a/packages/server-utils/src/utils/monitorStream.ts b/packages/server-utils/src/utils/monitorStream.ts new file mode 100644 index 000000000000..d19466a6a035 --- /dev/null +++ b/packages/server-utils/src/utils/monitorStream.ts @@ -0,0 +1,40 @@ +/** Monitor the source without reading ahead or treating cancellation as a source failure. */ +export function monitorStream( + stream: ReadableStream, + onDone: () => void, + onError: (error: unknown) => void, +): ReadableStream { + const reader = stream.getReader(); + // A cancelled reader closes normally; only a source failure should be captured as an error. + void reader.closed.then(onDone, error => { + onError(error); + onDone(); + }); + + return new ReadableStream( + { + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + reader.releaseLock(); + } else { + controller.enqueue(value); + } + } catch (error) { + controller.error(error); + reader.releaseLock(); + } + }, + async cancel(reason) { + try { + await reader.cancel(reason); + } finally { + reader.releaseLock(); + } + }, + }, + { highWaterMark: 0 }, + ); +} diff --git a/packages/server-utils/src/utils/wrapServerRequestHandler.ts b/packages/server-utils/src/utils/wrapServerRequestHandler.ts new file mode 100644 index 000000000000..9702af119181 --- /dev/null +++ b/packages/server-utils/src/utils/wrapServerRequestHandler.ts @@ -0,0 +1,270 @@ +import { + CLIENT_ADDRESS, + CLIENT_PORT, + HTTP_REQUEST_BODY_SIZE, + NETWORK_PROTOCOL_NAME, + SENTRY_OP, + SENTRY_SEGMENT_NAME_SOURCE, + USER_AGENT_ORIGINAL, +} from '@sentry/conventions/attributes'; +import { HTTP_SERVER } from '@sentry/conventions/op'; +import type { Client, Mechanism, Scope, Span } from '@sentry/core'; +import { + captureBodyFromWinterCGRequest, + captureException, + continueTrace, + getHttpSpanDetailsFromUrlObject, + hasSpansEnabled, + hasSpanStreamingEnabled, + httpHeadersToSpanAttributes, + HTTP_SPAN_NAME_FALLBACK, + parseStringToURLObject, + safeCallback, + setHttpStatus, + SPAN_STATUS_ERROR, + startSpanManual, + winterCGHeadersToDict, + winterCGRequestToRequestData, +} from '@sentry/core'; +import type { ServerRequestOptions } from '@sentry/core/server'; +import { + classifyResponseStreaming, + getClientIPAddress, + isStaticAssetRequest, + startRequestSession, +} from '@sentry/core/server'; +import { DEBUG_BUILD } from '../debug-build'; +import { monitorStream } from './monitorStream'; + +export interface ServerRequestHandlerConfig { + request: Request; + client: Client; + /** The isolation scope of this request. */ + isolationScope: Scope; + /** The options of the runtime's integration for incoming requests. */ + options: ServerRequestOptions | undefined; + /** Origin of the span, e.g. `auto.http.deno`. */ + origin: string; + /** Mechanism of errors that the handler or the response stream throws. */ + mechanism: Mechanism; + /** The address and port of the socket, if the runtime knows them. */ + getSocketAddress?: () => { address: string; port?: number } | undefined; +} + +/** + * Wraps a handler of a web `Request`, as used by `Deno.serve`, `Bun.serve` and Cloudflare Workers. + * + * It records the request session, attaches the request data and body, continues the incoming + * trace, creates the `http.server` span, and captures errors that the handler throws. The rules + * follow core's `node:http` server code: + * + * - `OPTIONS` and `HEAD` requests get no span and no body capture. + * - The static asset filter applies only to `GET`. + * - The span and the session of a streamed response end when the stream finishes. + */ +export async function wrapServerRequestHandler( + config: ServerRequestHandlerConfig, + handler: () => Response | Promise, +): Promise { + const { request, client, isolationScope, options, mechanism } = config; + const method = request.method.toUpperCase(); + + const endSession = options?.sessions + ? startRequestSession(client, { + requestIsolationScope: isolationScope, + sessionFlushingDelayMS: options.sessionFlushingDelayMS, + }) + : undefined; + + const socketAddress = config.getSocketAddress?.(); + isolationScope.setSDKProcessingMetadata({ + normalizedRequest: winterCGRequestToRequestData(request), + ipAddress: socketAddress?.address, + }); + + if (method === 'OPTIONS' || method === 'HEAD') { + try { + return await handler(); + } catch (e) { + captureException(e, { mechanism }); + throw e; + } finally { + endSession?.(); + } + } + + const dataCollection = client.getDataCollectionOptions(); + const effectiveBodySize = + options?.maxRequestBodySize ?? (dataCollection.httpBodies.includes('incomingRequest') ? 'medium' : 'none'); + const ignoreRequestBody = options?.ignoreRequestBody; + if ( + method !== 'GET' && + effectiveBodySize !== 'none' && + // If the filter throws, skip the body so it cannot leak data the filter was meant to hide. + !( + ignoreRequestBody && + safeCallback( + DEBUG_BUILD ? 'Error in `ignoreRequestBody`, skipping the body:' : '', + () => ignoreRequestBody(request.url, request), + () => true, + ) + ) + ) { + await captureBodyFromWinterCGRequest(request, isolationScope, effectiveBodySize); + } + + const urlObject = parseStringToURLObject(request.url); + const urlPath = urlObject ? `${urlObject.pathname}${urlObject.search}` : '/'; + const createSpan = shouldCreateSpan(client, options, request, urlPath); + + return continueTrace( + { + sentryTrace: request.headers.get('sentry-trace') || '', + baggage: request.headers.get('baggage'), + }, + async () => { + if (!createSpan) { + let response: Response; + try { + response = await handler(); + } catch (e) { + captureException(e, { mechanism }); + endSession?.(); + throw e; + } + isolationScope.setContext('response', { status_code: response.status }); + return onResponseDone(response, mechanism, () => endSession?.()); + } + + const [rawName, attributes] = getHttpSpanDetailsFromUrlObject( + urlObject, + 'server', + config.origin, + request, + undefined, + client, + ); + // With span streaming, span names have to be low cardinality, so we can't fall back to the URL. + // A `route` source means the name already is (e.g. the `/` path), so it is kept as-is. + const name = + attributes[SENTRY_SEGMENT_NAME_SOURCE] === 'route' || !hasSpanStreamingEnabled(client) + ? rawName + : method || HTTP_SPAN_NAME_FALLBACK; + + const contentLength = request.headers.get('content-length'); + if (contentLength) { + attributes[HTTP_REQUEST_BODY_SIZE] = parseInt(contentLength, 10); + } + const userAgent = request.headers.get('user-agent'); + if (userAgent) { + attributes[USER_AGENT_ORIGINAL] = userAgent; + } + + const headers = winterCGHeadersToDict(request.headers); + if (dataCollection.userInfo) { + // `client.address` is the originating client, so a forwarding header wins over the socket, which + // behind a proxy holds the proxy's address. The socket port is the proxy's too, so `client.port` + // stays unset then. + const forwardedAddress = getClientIPAddress(headers); + if (forwardedAddress) { + attributes[CLIENT_ADDRESS] = forwardedAddress; + } else if (socketAddress) { + attributes[CLIENT_ADDRESS] = socketAddress.address; + if (socketAddress.port !== undefined) { + attributes[CLIENT_PORT] = socketAddress.port; + } + } + } + + // describes the OSI application-layer protocol (http), not the scheme (might be https) + attributes[NETWORK_PROTOCOL_NAME] = 'http'; + Object.assign(attributes, httpHeadersToSpanAttributes(headers, dataCollection)); + attributes[SENTRY_OP] = HTTP_SERVER; + + return startSpanManual({ name, attributes }, async span => { + const { onSpanCreated, onSpanEnd } = options ?? {}; + if (onSpanCreated) { + safeCallback( + DEBUG_BUILD ? 'Error in `onSpanCreated`:' : '', + () => onSpanCreated(span, request), + () => undefined, + ); + } + + let response: Response; + try { + response = await handler(); + } catch (e) { + span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + span.end(); + captureException(e, { mechanism }); + endSession?.(); + throw e; + } + + setHttpStatus(span, response.status); + isolationScope.setContext('response', { status_code: response.status }); + span.setAttributes( + httpHeadersToSpanAttributes(winterCGHeadersToDict(response.headers), dataCollection, 'response'), + ); + + return onResponseDone( + response, + mechanism, + () => { + if (onSpanEnd) { + safeCallback( + DEBUG_BUILD ? 'Error in `onSpanEnd`:' : '', + () => onSpanEnd(span, request, response), + () => undefined, + ); + } + span.end(); + endSession?.(); + }, + span, + ); + }); + }, + ); +} + +function shouldCreateSpan( + client: Client, + options: ServerRequestOptions | undefined, + request: Request, + urlPath: string, +): boolean { + if (!(options?.spans ?? hasSpansEnabled(client.getOptions()))) { + return false; + } + if (options?.ignoreStaticAssets && request.method.toUpperCase() === 'GET' && isStaticAssetRequest(urlPath)) { + return false; + } + const ignoreIncomingRequests = options?.ignoreIncomingRequests; + return !( + ignoreIncomingRequests && + safeCallback( + DEBUG_BUILD ? 'Error in `ignoreIncomingRequests`, creating the span:' : '', + () => ignoreIncomingRequests(urlPath, request), + () => false, + ) + ); +} + +/** + * Calls `onDone` when the response is done: at once, or when a streamed body finishes. + * Returns the response to send. + */ +function onResponseDone(response: Response, mechanism: Mechanism, onDone: () => void, span?: Span): Response { + if (response.body && !response.body.locked && classifyResponseStreaming(response).isStreaming) { + const body = monitorStream(response.body, onDone, error => { + span?.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); + captureException(error, { mechanism }); + }); + // Passing the original response as the init keeps its status, status text and headers. + return new Response(body, response); + } + onDone(); + return response; +} diff --git a/packages/server-utils/test/utils/monitorStream.test.ts b/packages/server-utils/test/utils/monitorStream.test.ts new file mode 100644 index 000000000000..eea047f4cee8 --- /dev/null +++ b/packages/server-utils/test/utils/monitorStream.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, it, vi } from 'vitest'; +import { monitorStream } from '../../src/utils/monitorStream'; + +const encode = (text: string): Uint8Array => new TextEncoder().encode(text); + +// Runs every pending promise job, for checks that something did not happen. +const flushPromiseJobs = (): Promise => new Promise(resolve => setImmediate(resolve)); + +describe('monitorStream', () => { + it('passes the chunks through and calls onDone when the source closes', async () => { + const onDone = vi.fn(); + const onError = vi.fn(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encode('a')); + controller.enqueue(encode('b')); + controller.close(); + }, + }); + + const text = await new Response(monitorStream(source, onDone, onError)).text(); + + expect(text).toBe('ab'); + expect(onDone).toHaveBeenCalledTimes(1); + expect(onError).not.toHaveBeenCalled(); + }); + + it('reads from the source only when the consumer pulls', async () => { + let pulls = 0; + const source = new ReadableStream( + { + pull(controller) { + pulls++; + controller.enqueue(encode(String(pulls))); + }, + }, + { highWaterMark: 0 }, + ); + + const reader = monitorStream(source, vi.fn(), vi.fn()).getReader(); + await flushPromiseJobs(); + expect(pulls).toBe(0); + + await reader.read(); + await flushPromiseJobs(); + expect(pulls).toBe(1); + await reader.cancel(); + }); + + it('cancels the source when the consumer cancels, without reporting an error', async () => { + let sourceCancelReason: unknown; + const onDone = vi.fn(); + const onError = vi.fn(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encode('first event')); + }, + cancel(reason) { + sourceCancelReason = reason; + }, + }); + const reader = monitorStream(source, onDone, onError).getReader(); + const reason = new Error('client disconnected'); + + await reader.read(); + await reader.cancel(reason); + await vi.waitFor(() => expect(onDone).toHaveBeenCalledTimes(1)); + + expect(sourceCancelReason).toBe(reason); + expect(onError).not.toHaveBeenCalled(); + }); + + it('reports a source error and calls onDone, without an unhandled rejection', async () => { + const unhandled = vi.fn(); + process.on('unhandledRejection', unhandled); + const onDone = vi.fn(); + const onError = vi.fn(); + const error = new Error('source failed'); + let controller: ReadableStreamDefaultController; + const source = new ReadableStream({ + start(value) { + controller = value; + }, + }); + + try { + const response = new Response(monitorStream(source, onDone, onError)); + const text = response.text(); + controller!.error(error); + + await expect(text).rejects.toBe(error); + await vi.waitFor(() => expect(onDone).toHaveBeenCalledTimes(1)); + await flushPromiseJobs(); + + expect(onError).toHaveBeenCalledWith(error); + expect(unhandled).not.toHaveBeenCalled(); + } finally { + process.off('unhandledRejection', unhandled); + } + }); +}); diff --git a/packages/server-utils/test/utils/wrapServerRequestHandler.test.ts b/packages/server-utils/test/utils/wrapServerRequestHandler.test.ts new file mode 100644 index 000000000000..4acd34c92bbc --- /dev/null +++ b/packages/server-utils/test/utils/wrapServerRequestHandler.test.ts @@ -0,0 +1,157 @@ +import type { Event } from '@sentry/core'; +import { getIsolationScope, setCurrentClient, withIsolationScope } from '@sentry/core'; +import { beforeAll, describe, expect, it, vi } from 'vitest'; +import { setAsyncLocalStorageAsyncContextStrategy } from '../../src/async-context'; +import type { ServerRequestHandlerConfig } from '../../src/utils/wrapServerRequestHandler'; +import { wrapServerRequestHandler } from '../../src/utils/wrapServerRequestHandler'; +import { getDefaultTestClientOptions, TestClient } from '../mocks/client'; + +const MECHANISM = { handled: false, type: 'auto.http.test' }; + +function setup(options: ServerRequestHandlerConfig['options'] = {}): { + events: Event[]; + transactions: Event[]; + serve: (request: Request, handler: () => Response | Promise) => Promise; +} { + const events: Event[] = []; + const transactions: Event[] = []; + const client = new TestClient( + getDefaultTestClientOptions({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + tracesSampleRate: 1, + beforeSend: event => { + events.push(event); + return null; + }, + beforeSendTransaction: event => { + transactions.push(event); + return null; + }, + }), + ); + setCurrentClient(client); + client.init(); + + const serve = (request: Request, handler: () => Response | Promise): Promise => + withIsolationScope(isolationScope => + wrapServerRequestHandler( + { request, client, isolationScope, options, origin: 'auto.http.test', mechanism: MECHANISM }, + handler, + ), + ); + return { events, transactions, serve }; +} + +describe('wrapServerRequestHandler', () => { + beforeAll(() => { + setAsyncLocalStorageAsyncContextStrategy(); + }); + + it.each(['OPTIONS', 'HEAD'])( + 'attaches request data and captures errors for %s requests, without a span', + async method => { + const { events, transactions, serve } = setup(); + const error = new Error(`${method} failed`); + + await expect( + serve(new Request('https://example.com/cors', { method }), () => { + expect(getIsolationScope().getScopeData().sdkProcessingMetadata.normalizedRequest?.method).toBe(method); + throw error; + }), + ).rejects.toBe(error); + + expect(events).toHaveLength(1); + expect(events[0]?.exception?.values?.[0]?.mechanism).toEqual(MECHANISM); + expect(transactions).toHaveLength(0); + }, + ); + + it('skips the body when ignoreRequestBody throws', async () => { + const { serve } = setup({ + ignoreRequestBody: () => { + throw new Error('filter failed'); + }, + }); + let data: unknown = 'not read'; + + const response = await serve(new Request('https://example.com/login', { method: 'POST', body: 'secret' }), () => { + data = getIsolationScope().getScopeData().sdkProcessingMetadata.normalizedRequest?.data; + return new Response('ok'); + }); + + expect(await response.text()).toBe('ok'); + expect(data).toBeUndefined(); + }); + + it('calls onSpanEnd and ends the span when a streamed response finishes', async () => { + const calls: string[] = []; + const { transactions, serve } = setup({ + onSpanEnd: (span, _request, response) => { + calls.push('onSpanEnd'); + span.setAttribute('onSpanEnd.status', response.status); + }, + }); + let controller: ReadableStreamDefaultController; + const source = new ReadableStream({ + start(value) { + controller = value; + }, + }); + + const response = await serve( + new Request('https://example.com/events'), + () => new Response(source, { status: 201, headers: { 'content-type': 'text/event-stream' } }), + ); + expect(response.status).toBe(201); + expect(response.headers.get('content-type')).toBe('text/event-stream'); + controller!.enqueue(new TextEncoder().encode('data: one\n\n')); + expect(calls).toEqual([]); + + controller!.close(); + expect(await response.text()).toBe('data: one\n\n'); + await vi.waitFor(() => expect(transactions).toHaveLength(1)); + + expect(calls).toEqual(['onSpanEnd']); + expect(transactions[0]?.contexts?.trace?.data?.['onSpanEnd.status']).toBe(201); + }); + + // Serves a streamed response whose source fails after the handler returns. + async function serveFailingStream(serve: ReturnType['serve']): Promise { + const error = new Error('stream interrupted'); + let controller: ReadableStreamDefaultController; + const source = new ReadableStream({ + start(value) { + controller = value; + }, + }); + + const response = await serve( + new Request('https://example.com/events'), + () => new Response(source, { headers: { 'content-type': 'application/x-ndjson' } }), + ); + const text = response.text(); + controller!.error(error); + await expect(text).rejects.toBe(error); + } + + it('captures an error from a streamed response source and marks the span as failed', async () => { + const { events, transactions, serve } = setup({ spans: true }); + + await serveFailingStream(serve); + await vi.waitFor(() => expect(transactions).toHaveLength(1)); + + expect(events).toHaveLength(1); + expect(events[0]?.exception?.values?.[0]?.mechanism).toEqual(MECHANISM); + expect(transactions[0]?.contexts?.trace?.status).toBe('internal_error'); + }); + + it('captures an error from a streamed response source without a span', async () => { + const { events, transactions, serve } = setup({ spans: false }); + + await serveFailingStream(serve); + await vi.waitFor(() => expect(events).toHaveLength(1)); + + expect(events[0]?.exception?.values?.[0]?.mechanism).toEqual(MECHANISM); + expect(transactions).toHaveLength(0); + }); +});