diff --git a/CHANGELOG.md b/CHANGELOG.md index b37287670bfc..e93885103e3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -242,6 +242,14 @@ Work in this release was contributed by @4ktLuffy, @adityareddy-dev, @dasilvalui 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); + }); +});