Node.js 流式导出中途失败：用 pipeline 传递错误并清理

报表下载中途写入失败，如果调用方只等源流读完，就可能提前记录成功。用 Promise 版 pipeline 等整个链路结束，把写入结果作为发布条件，并清理失败后留下的半成品。

1. 模拟写入第三块时失败

保存为 stream_lab.mjs，运行 node stream_lab.mjs。Writable 用回调模拟慢写端，第二次运行在第三块报错。

import assert from 'node:assert/strict';
import { Readable, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

async function runExport(failAt) {
  let accepted = 0;
  let finalCalled = false;
  const source = Readable.from(['one', 'two', 'three', 'four']);
  const sink = new Writable({
    highWaterMark: 3,
    write(chunk, encoding, callback) {
      setTimeout(() => {
        if (accepted + 1 === failAt) {
          callback(new Error('injected write failure'));
          return;
        }
        accepted++;
        callback();
      }, 5);
    },
    final(callback) {
      finalCalled = true;
      callback();
    },
  });
  if (failAt) {
    await assert.rejects(pipeline(source, sink), /injected write failure/);
    assert.equal(accepted, 2);
    assert.equal(finalCalled, false);
    assert.equal(source.destroyed, true);
    assert.equal(sink.destroyed, true);
    console.log('failure: accepted=2, final=false, streams destroyed');
  } else {
    await pipeline(source, sink);
    assert.equal(accepted, 4);
    assert.equal(finalCalled, true);
    console.log('success: accepted=4, final=true');
  }
}

await runExport(0);
await runExport(3);

本地实测输出如下。

success: accepted=4, final=true
failure: accepted=2, final=false, streams destroyed

2. 把写入完成和发布分成两步

写文件时让 pipeline 输出到同目录临时路径，成功后再重命名，失败时删除临时文件。直接流式返回 HTTP 也能传播失败，但已发送的响应头无法改写为 500，客户端应处理下载中断。需要让用户只看到完整导出时，先生成产物，再提供下载地址。

highWaterMark 是触发背压的阈值。一个大块仍可能超过阈值，变换流两端也各有缓冲。要给内存估算上限，还要检查块大小、压缩器缓冲和并发导出数。若 Transform 内部先把全量数据收集到数组，外部 pipeline 无法为这部分提供背压。

小结

等待 pipeline 完成后才记录导出成功。查失败时同时看最后成功写入的块、目标端错误，以及临时文件的清理结果。

数据来源与检索时间

检索时间 2026-11-13。

核验说明

• Node.js v24.19.0 验证成功与注入失败。写端为内存模拟，磁盘替换、网络传输和内存峰值属于部署场景。