Close the streams you hand to other code

const handedOut = new AbortController()
try {
  for (const stage of pipeline) {
    const src = new Response(body.pipeThrough(new TransformStream(), { signal: handedOut.signal }))
    body = (await stage(src)).body
  }
} catch (err) {
  handedOut.abort(err)                 // closes bodies a stage locked, then threw
  if (!body?.locked) await body?.cancel().catch(() => {})
  throw err
}