node:http client span never ends when the response is destroyed without an error (res.destroy()) #25469

// Repro: http.client span is never ended when the response is destroyed mid-body.
// Usage: npm i @sentry/node && node nodehttp-repro.mjs
// Everything is local (127.0.0.1); no real DSN is used.
import * as Sentry from '@sentry/node';
import http from 'node:http';

// Local DSN sink: collects envelopes so we can see which spans were actually sent.
const sentSpans = new Map(); // url.full -> status, for http.client spans that reached the sink
const collect = o => {
  if (!o || typeof o !== 'object') return;
  const a = o.attributes || {};
  const url = a['url.full']?.value ?? a['url.full'] ?? o.data?.['url.full'];
  const op = a['sentry.op']?.value ?? a['sentry.op'] ?? o.op;
  if (op === 'http.client' && url) sentSpans.set(url, o.status);
  for (const v of Object.values(o)) if (v && typeof v === 'object') collect(v);
};
const sink = http.createServer((req, res) => {
  let body = '';
  req.on('data', c => (body += c));
  req.on('end', () => {
    for (const line of body.split('\n')) { try { collect(JSON.parse(line)); } catch {} }
    res.end('{}');
  });
});
await new Promise(r => sink.listen(0, '127.0.0.1', r));

Sentry.init({
  dsn: `http://k@127.0.0.1:${sink.address().port}/1`,
  tracesSampleRate: 1,
  // do not trace the sink itself
  ignoreOutgoingRequests: url => url.includes(`:${sink.address().port}`),
});

const ended = new Map(); // url.full -> status
Sentry.getClient().on('spanEnd', span => {
  const j = Sentry.spanToJSON(span);
  const url = j.attributes?.['url.full'] ?? j.data?.['url.full'];
  const op = j.attributes?.['sentry.op'] ?? j.op;
  if (op === 'http.client' && url) ended.set(url, j.status);
});

// Server that streams a slow body (headers sent, body never finishes on its own).
const srv = http.createServer((req, res) => {
  res.writeHead(200);
  res.write('first chunk');
  if (req.url.startsWith('/full')) setTimeout(() => res.end('rest'), 50);
});
await new Promise(r => srv.listen(0, '127.0.0.1', r));
const port = srv.address().port;

const cases = {
  full: res => { res.on('data', () => {}); },
  'res.destroy()': res => { res.once('data', () => res.destroy()); },
  'res.destroy(new Error())': res => { res.once('data', () => res.destroy(new Error('boom'))); },
};

await Sentry.startSpan({ name: 'root' }, async () => {
  for (const [name, handle] of Object.entries(cases)) {
    await new Promise(resolve => {
      const path = `/${name === 'full' ? 'full' : 'slow'}?case=${encodeURIComponent(name)}`;
      const req = http.get({ host: '127.0.0.1', port, path }, res => {
        handle(res);
        res.on('close', () => setTimeout(resolve, 300));
        res.on('error', () => {});
      });
      req.on('error', () => {});
    });
  }
});

await Sentry.flush(2000);
for (const name of Object.keys(cases)) {
  const path = `/${name === 'full' ? 'full' : 'slow'}?case=${encodeURIComponent(name)}`;
  const url = `http://127.0.0.1:${port}${path}`;
  console.log(
    name.padEnd(26),
    '| spanEnd fired:', ended.has(url) ? `yes (status=${ended.get(url)})` : 'NO',
    '| span in envelope:', sentSpans.has(url) ? `yes (status=${sentSpans.get(url)})` : 'NO',
  );
}
srv.closeAllConnections();
srv.close();
sink.close();
setTimeout(() => process.exit(0), 100);