From e604efafd260c730412c5a4565c77a62a542d951 Mon Sep 17 00:00:00 2001 From: breken Date: Sun, 13 Sep 2026 19:25:43 -0700 Subject: [PATCH] fix(http): treat end(callback) argument as a callback, not body data Node's documented signatures are response.end([data[, encoding]][, callback]) and request.end([data[, encoding]][, callback]). The record-mode patches on ServerResponse.end and ClientRequest.end only checked `if (chunk)`, so a callback-only end() pushed the function into the body chunk list. Server side, combineChunks() then threw ERR_INVALID_ARG_TYPE inside the nextTick body processing; the error was logged and the span was exported without its response body, even when res.write() had already sent one. Client side, Buffer.from(callback, encoding) threw synchronously out of the user's own req.end(cb) call, so the outbound request never left the process while recording was on. Guard both patches (and the identical ServerResponse.end patch in the Next.js instrumentation) with typeof chunk !== "function" and pass the argument through to the original end() unchanged, which already handles the callback form. --- .../libraries/http/Instrumentation.ts | 6 +- .../integration-tests/end-callback.test.ts | 176 ++++++++++++++++++ .../libraries/nextjs/Instrumentation.ts | 3 +- 3 files changed, 182 insertions(+), 3 deletions(-) create mode 100644 src/instrumentation/libraries/http/integration-tests/end-callback.test.ts diff --git a/src/instrumentation/libraries/http/Instrumentation.ts b/src/instrumentation/libraries/http/Instrumentation.ts index 2478430b..2d602845 100644 --- a/src/instrumentation/libraries/http/Instrumentation.ts +++ b/src/instrumentation/libraries/http/Instrumentation.ts @@ -413,7 +413,8 @@ export class HttpInstrumentation extends TdInstrumentationBase { const statusCode = res.statusCode; // Capture any final response data - if (chunk) { + // res.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function") { responseChunks.push(chunk); } @@ -755,7 +756,8 @@ export class HttpInstrumentation extends TdInstrumentationBase { if (originalEnd) { req.end = function (chunk?: any, encoding?: any, callback?: any): ClientRequest { // Capture final chunk if provided - if (chunk && !requestBodyCaptured) { + // req.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function" && !requestBodyCaptured) { requestBodyChunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding)); } diff --git a/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts b/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts new file mode 100644 index 00000000..f743e0ba --- /dev/null +++ b/src/instrumentation/libraries/http/integration-tests/end-callback.test.ts @@ -0,0 +1,176 @@ +process.env.TUSK_DRIFT_MODE = "RECORD"; + +import { TuskDrift } from "../../../../core/TuskDrift"; + +TuskDrift.initialize({ + apiKey: "test-api-key-end-callback", + env: "test", + logLevel: "silent", +}); +TuskDrift.markAppAsReady(); + +import { SpanKind } from "@opentelemetry/api"; +import { + InMemorySpanAdapter, + registerInMemoryAdapter, + clearRegisteredInMemoryAdapters, +} from "../../../../core/tracing/adapters/InMemorySpanAdapter"; +import { waitForSpans } from "./test-utils"; +import test from "ava"; +import type { Server, RequestOptions } from "http"; +import { AddressInfo } from "net"; + +const spanAdapter = new InMemorySpanAdapter(); +registerInMemoryAdapter(spanAdapter); + +// Require http after TuskDrift.initialize() so the require-in-the-middle hook patches it, +// matching how the SDK is used in an application. +const http: typeof import("http") = require("http"); + +let server: Server; +let port: number; + +function decodeBody(value: unknown): string { + return Buffer.from(String(value), "base64").toString("utf8"); +} + +function makeRequest( + options: RequestOptions, + body?: string, + endWithCallback = false, +): Promise<{ statusCode: number; data: string }> { + return new Promise((resolve, reject) => { + const req = http.request(options, (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => resolve({ statusCode: res.statusCode || 0, data })); + }); + req.on("error", reject); + if (body) { + req.write(body); + } + if (endWithCallback) { + // Node's documented signature: request.end([data[, encoding]][, callback]) + req.end(() => {}); + } else { + req.end(); + } + }); +} + +test.before(async () => { + server = http.createServer((req, res) => { + if (req.url === "/upstream") { + let body = ""; + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ upstreamReceived: body })); + }); + return; + } + + if (req.url === "/call-upstream") { + // Outbound request made inside the inbound span, ended with a callback-only end(). + const postData = JSON.stringify({ hello: "world" }); + try { + makeRequest( + { + hostname: "127.0.0.1", + port, + path: "/upstream", + method: "POST", + headers: { + "content-type": "application/json", + "content-length": Buffer.byteLength(postData), + }, + }, + postData, + true, + ).then( + (upstream) => { + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ upstream: JSON.parse(upstream.data) })); + }, + (error: Error) => { + res.statusCode = 500; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ error: error.message })); + }, + ); + } catch (error) { + res.statusCode = 500; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ error: (error as Error).message })); + } + return; + } + + // Inbound handler that writes the body, then ends with a callback-only end(). + res.setHeader("content-type", "application/json"); + res.write(JSON.stringify({ path: req.url })); + res.end(() => {}); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + port = (server.address() as AddressInfo).port; +}); + +test.after.always(async () => { + await new Promise((resolve, reject) => + server.close((err?: Error) => (err ? reject(err) : resolve())), + ); + clearRegisteredInMemoryAdapters(); +}); + +test("inbound: response body is recorded when the handler calls res.end(callback)", async (t) => { + const response = await makeRequest({ + hostname: "127.0.0.1", + port, + path: "/inbound", + method: "GET", + }); + + t.is(response.statusCode, 200); + t.deepEqual(JSON.parse(response.data), { path: "/inbound" }); + + await waitForSpans(); + + const serverSpan = spanAdapter + .getAllSpans() + .find( + (span) => span.kind === SpanKind.SERVER && (span.inputValue as any)?.target === "/inbound", + ); + t.truthy(serverSpan, "Expected a SERVER span for /inbound"); + + const outputValue = serverSpan!.outputValue as any; + t.truthy(outputValue.body, "Expected the recorded server span to carry the response body"); + t.deepEqual(JSON.parse(decodeBody(outputValue.body)), { path: "/inbound" }); +}); + +test("outbound: req.end(callback) inside a trace does not throw and the request body is recorded", async (t) => { + const response = await makeRequest({ + hostname: "127.0.0.1", + port, + path: "/call-upstream", + method: "GET", + }); + + t.is(response.statusCode, 200, `Unexpected response: ${response.data}`); + t.deepEqual(JSON.parse(response.data), { + upstream: { upstreamReceived: JSON.stringify({ hello: "world" }) }, + }); + + await waitForSpans(); + + const clientSpan = spanAdapter + .getAllSpans() + .find( + (span) => span.kind === SpanKind.CLIENT && (span.inputValue as any)?.path === "/upstream", + ); + t.truthy(clientSpan, "Expected a CLIENT span for /upstream"); + + const inputValue = clientSpan!.inputValue as any; + t.truthy(inputValue.body, "Expected the recorded client span to carry the request body"); + t.deepEqual(JSON.parse(decodeBody(inputValue.body)), { hello: "world" }); +}); diff --git a/src/instrumentation/libraries/nextjs/Instrumentation.ts b/src/instrumentation/libraries/nextjs/Instrumentation.ts index 23202a4f..760a659c 100644 --- a/src/instrumentation/libraries/nextjs/Instrumentation.ts +++ b/src/instrumentation/libraries/nextjs/Instrumentation.ts @@ -352,7 +352,8 @@ export class NextjsInstrumentation extends TdInstrumentationBase { const originalEnd = actualRes.end?.bind(actualRes); if (originalEnd) { actualRes.end = function (chunk?: any, encoding?: any, callback?: any) { - if (chunk) { + // res.end(callback) passes the callback as the first argument; it is not body data + if (chunk && typeof chunk !== "function") { responseChunks.push(chunk); } return originalEnd.call(this, chunk, encoding, callback);