HTTP stream append retry after 429 reuses a consumed fetch body #4628

import assert from 'node:assert/strict';
import { createServer } from 'node:http';
import { createRequire } from 'node:module';
import path from 'node:path';
import { pathToFileURL } from 'node:url';

const require = createRequire(path.resolve('package.json'));
const { createStreamer } = await import(
  pathToFileURL(require.resolve('@workflow/world-vercel')).href
);
const payload = new TextEncoder().encode('Synthetic caf\u00e9 \ud83d\ude00 bytes.');
const attempts = new Map();
const received = [];
const committed = new Map();
const server = createServer((req, res) => {
  const name = new URL(req.url, 'http://localhost').pathname.split('/').at(-1);
  const attempt = (attempts.get(name) ?? 0) + 1;
  attempts.set(name, attempt);
  req.on('error', () => {});
  const pieces = [];
  req.on('data', piece => pieces.push(piece));
  req.on('end', () => {
    const bytes = Buffer.concat(pieces);
    received.push({ name, attempt, bytes: bytes.length });
    if (name === 'strm_retry' && attempt === 1) return;
    committed.set(name, [...(committed.get(name) ?? []), bytes]);
    res.writeHead(200).end('ok');
  });
  // Reject before consuming the body. This request is never committed.
  if (name === 'strm_retry' && attempt === 1) {
    res.writeHead(429, { 'Retry-After': '0.001' }).end('rate limited');
  }
});

const originalEnv = { ...process.env };
try {
  await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
  process.env.VERCEL_WORKFLOW_SERVER_URL = `http://127.0.0.1:${server.address().port}`;
  process.env.WORKFLOW_NODE_HTTP = '0';
  const streamer = createStreamer();
  await streamer.streams.write('wrun_example', 'strm_control', payload);
  assert.deepEqual(committed.get('strm_control'), [Buffer.from(payload)]);
  let failure;
  try {
    await streamer.streams.write('wrun_example', 'strm_retry', payload);
  } catch (error) {
    failure = error;
  }
  const causes = [];
  const seen = new Set();
  for (let error = failure; error && !seen.has(error); error = error.cause) {
    seen.add(error);
    causes.push({ name: error.name, code: error.code });
  }
  console.log(JSON.stringify({ received, attempts: Object.fromEntries(attempts), causes }));
  // Fails on unpatched 5.0.1 with UND_ERR_REQ_CONTENT_LENGTH_MISMATCH.
  assert.equal(failure, undefined, 'Confirmed 429 should retry a complete body');
  assert.equal(attempts.get('strm_retry'), 2);
  assert.deepEqual(committed.get('strm_retry'), [Buffer.from(payload)]);
  console.log('PASS: healthy control and exactly one complete append after 429');
} finally {
  for (const key of ['VERCEL_WORKFLOW_SERVER_URL', 'WORKFLOW_NODE_HTTP']) {
    if (originalEnv[key] === undefined) delete process.env[key];
    else process.env[key] = originalEnv[key];
  }
  server.closeAllConnections();
  await new Promise(resolve => server.close(resolve));
}