# 文件、流与上传

## 本章解决什么问题

上传一个 PDF 后，页面可能只看到一次成功提示，服务端却经历了接收、写入、校验、解析和登记状态。直接把整个文件读进内存，再交给解析器，少量文件时没有问题，并发稍高就可能挤满内存；失败留下的半截文件又会被后续任务误认为完整文档。本章目标是理解 Buffer、流、背压与文件生命周期，能实现有限大小、失败可清理、成功后才发布的文件接收核心。前置是取消信号与 HTTP 请求体读取。

前端熟悉 Blob、File 和 FormData，但它们只是浏览器侧的数据表示与封装。服务端收到的是网络字节，客户端提供的文件名和 Content-Type 都只是声明，不是事实。把 `../../config.json` 当成本地路径，或把“声称是 PDF”的可执行内容直接公开，都是错误的信任方向。

## 从整块数据到有背压的流水线

Buffer 表示字节序列；字符串表示解码后的文本。一个汉字通常占多个 UTF-8 字节，任意一个网络块可能切在字符编码中间。不能把每个块单独转为字符串后直接拼接，否则边界处可能出现替代字符；处理完整小文本可以先拼 Buffer 再解码，处理大文本可用流式解码器。

可读流提供数据，可写流消费数据，Transform 位于中间并改变或检查数据。想象前端虚拟列表：你不会一次把百万行都塞进 DOM；文件处理也应该让下游速度决定上游节奏。背压就是下游暂时处理不完时，上游减慢供给。它不能让磁盘无限快，但能防止生产速度持续超过消费速度时，内存无限累积。

`pipeline` 把流连接起来，处理背压并传播错误。只是手写 `readable.on('data', chunk => writable.write(chunk))` 却忽略 write 的返回值，会破坏背压。当链路任一环失败时，还需处理流以外的业务资源，比如删除临时文件、撤销待解析状态；pipeline 不知道你的数据库记录代表什么。[Node Streams](https://nodejs.org/api/stream.html) 给出了流组合与错误行为的完整定义。

## 核心 API 合同

`createWriteStream(path, { flags: 'wx', mode: 0o600 })` 返回可写流；wx 表示以排他创建方式打开，目标已存在会报错。mode 在支持 POSIX 权限的环境有意义，Windows 的实际访问控制还取决于 ACL，不能把这一位参数当成跨平台隔离方案。

`new Transform({ transform(chunk, encoding, callback) {} })` 收到一块数据，调用 callback(null, chunk) 继续传递，或 callback(error) 终止。每块必须恰好回调一次。Promise 版 `pipeline(source, transform, destination, { signal })` 成功时兑现，流错误或取消时拒绝；失败会尝试销毁链路，因此直接把 HTTP IncomingMessage 放进失败的 pipeline 时，客户端可能只看到连接断开，不能承诺总能收到整齐的 JSON 413。

`rename(oldPath, newPath)` 返回 Promise；示例让临时文件和最终文件位于同一个目录，避免跨文件系统移动。`unlink(path)` 删除一个文件，目标不存在会报 ENOENT；清理时只可有意识忽略这个错误，不应吞掉权限拒绝或磁盘异常。[Node File System](https://nodejs.org/api/fs.html) 描述了这些操作的返回与错误。

## 完整示例：安全落盘核心与超限清理

环境：Node 22.22 或 24，无依赖。保存为 `save-stream.mjs`，执行 `node save-stream.mjs`。程序在系统临时目录中创建专属练习目录，只处理自己生成的数据，验收后删除自己创建的文件和目录。它演示上传接收的存储核心，不是 multipart 解析器或完整 HTTP 上传服务。

```js save-stream.mjs
// save-stream.mjs
import { Readable, Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createWriteStream } from 'node:fs';
import { mkdtemp, rename, unlink, readdir, rmdir, readFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { randomUUID } from 'node:crypto';
import assert from 'node:assert/strict';

async function removeIfPresent(path) {
  try { await unlink(path); }
  catch (error) { if (error.code !== 'ENOENT') throw error; }
}

async function saveStream(source, directory, { maxBytes, signal } = {}) {
  if (!Number.isSafeInteger(maxBytes) || maxBytes < 1) {
    throw new RangeError('maxBytes 必须为正安全整数');
  }
  // 路径完全由服务端生成，不拼接用户上传的文件名。
  const id = randomUUID();
  const temporary = join(directory, `${id}.part`);
  const finalPath = join(directory, `${id}.bin`);
  let bytes = 0;
  const guard = new Transform({
    transform(chunk, encoding, callback) {
      bytes += chunk.length;
      if (bytes > maxBytes) {
        callback(Object.assign(new Error('文件超过大小上限'), { code: 'TOO_LARGE' }));
        return;
      }
      callback(null, chunk);
    },
  });

  try {
    await pipeline(source, guard,
      createWriteStream(temporary, { flags: 'wx', mode: 0o600 }),
      { signal });
    // 整条流成功结束后才出现最终文件名。
    await rename(temporary, finalPath);
    return { id, bytes, path: finalPath };
  } catch (error) {
    await removeIfPresent(temporary);
    throw error;
  }
}

const directory = await mkdtemp(join(tmpdir(), 'node-stream-lesson-'));
try {
  const file = await saveStream(
    Readable.from([Buffer.from('知识'), Buffer.from('库')]),
    directory,
    { maxBytes: 20, signal: AbortSignal.timeout(3000) },
  );
  assert.equal(file.bytes, 9);
  assert.equal(await readFile(file.path, 'utf8'), '知识库');
  console.log('成功写入 9 字节');

  await assert.rejects(() => saveStream(
    Readable.from([Buffer.alloc(8), Buffer.alloc(8)]),
    directory, { maxBytes: 10 },
  ), { code: 'TOO_LARGE' });
  const names = await readdir(directory);
  assert.equal(names.length, 1);
  assert.ok(names.every((name) => !name.endsWith('.part')));
  console.log('超限任务失败，未留下临时文件');
} finally {
  // 目录由本程序生成且不接受外部路径，只删除自己的练习文件。
  for (const name of await readdir(directory)) await unlink(join(directory, name));
  await rmdir(directory);
}
```

预期输出两行：成功写入九字节、超限任务失败且无临时文件。路径与 UUID 每次变化，示例不把它们视为固定结果。正式项目中的持久目录必须由配置提供，并有容量与权限管理，不能长期把业务资料寄存在系统临时目录。

## 关键代码逐段说明

路径用随机标识生成，原始文件名可作为经过长度限制的元数据保存，但不参与本地路径拼接。临时文件与最终文件分离，使下游只处理已完成的对象；若在接收过程中就创建“已上传”状态，解析任务可能读到半截内容。数据库状态更新与对象存储操作无法天然共享一个事务，需要设计 uploading、ready、failed 等状态及补偿清理。

guard 在每块进入磁盘前检查累计字节，这比只校验客户端 Content-Length 可靠。它仍不能限制压缩内容解压后的体积，也不能证明文件格式正确。PDF 解析、图片解码、压缩包解压应各自设置页数、像素数、展开体积、CPU 时间等限制；上传大小只是第一道界限。

pipeline 成功后 rename，把“写完”与“可被使用”连接起来。同一文件系统的重命名通常适合作为可见性切换，但不等于已经满足所有断电持久性要求，也不解决随机名字极端碰撞和跨系统一致性。需要更严格保证时应采用对象存储的对象标识、校验和与明确提交步骤。

## 从教学核心接到真实上传

multipart/form-data 包含边界、字段与多份内容，应使用维护中的流式解析器，并同时限制文件数、字段数、字段长度和总大小。大文件可以采用受限的预签名上传流程，让浏览器直接传对象存储；后端仍需验证上传完成对象的大小、类型、租户归属和有效期，不能收到一个 URL 就认为它属于当前用户。

下载也要重新做资源归属检查。不要把私有文档放入无需鉴权的静态目录，不能只依赖随机文件名保密。保存 HTML、SVG 或可执行内容时，还要考虑浏览器打开后的脚本行为；业务不需要在线预览时，可通过下载响应与隔离域减少风险。[OWASP 文件上传建议](https://cheatsheetseries.owasp.org/cheatsheets/File_Upload_Cheat_Sheet.html) 提供了上传检查维度。

## 从上传按钮看到完整的文档生命周期

“团队文档与任务助手”接收文件以后，至少要知道上传是否完整、对象是否存在、格式能否解析、文本是否提取成功、索引是否发布。一个 uploaded 布尔值很难同时表达这些事实。更清楚的模型是把文件接收状态与解析任务状态分开，接收完成才允许领取解析任务，索引成功后才把新版本展示为可检索。

前端进度条达到百分之百，通常只意味着浏览器已经发送了请求体，不代表服务端已校验、写入持久存储或建立索引。界面可以显示“上传完成，正在处理”，并使用任务编号查询后续进度。这个区别能避免用户看到成功提示后立刻搜索，却因为后台仍在解析而以为资料丢失。

文件本身与文档实体也应分开。一个文档可对应多个上传版本，一个对象可能保存原始字节，一个解析版本又生成多个片段和缩略图。把这些都塞进单个 file_path 字段，会在重新上传、失败重试和历史恢复时失去关系。每个产物应知道来源版本、大小、校验摘要、存储位置和所属租户。

## 字节与字符：为什么分块边界不能随意解码

UTF-8 中一个字符可能占多个字节，网络与磁盘读取块按缓冲与传输条件形成，不会照顾字符边界。把每个 Buffer 单独 toString 后拼接，会把被切开的编码序列分别替换成无效字符。英文样本可能永远发现不了这个问题，因此团队文档助手的最小测试应包含中文和多字节符号，而不是只有 hello。

`StringDecoder(encoding)` 默认采用 UTF-8，write(buffer) 返回当前能够完整解码的文本，并暂存末尾未完成的字节；end() 返回剩余内容并结束解码。它适合保持流式文本边界，但并不是严格的文件有效性验证器，损坏编码可能被替换。要求拒绝无效编码时，应采用有明确严格模式的解码方式，并决定对 BOM、换行和混合编码的策略。

Buffer.byteLength(text,'utf8') 返回编码后的字节数，字符串 length 反映 UTF-16 码元数，两者服务不同目的。网络限制用字节，用户标题限制可能用码元、码点或感知字符；不要在前端按一种方式验证，后端按另一种方式悄悄截断。二进制 PDF 更不应先转成字符串再保存，否则原始字节可能被破坏。

## 最小实验：一个汉字被分成三块

保存为 `utf8-stream.mjs`，Node 22.22 或 24 执行 `node utf8-stream.mjs`，无依赖。代码故意每次只提供一个字节，展示逐块解码与有状态解码的差异。

```js utf8-stream.mjs
// utf8-stream.mjs
import { StringDecoder } from 'node:string_decoder';
import assert from 'node:assert/strict';
const original='团队文档';
const bytes=Buffer.from(original,'utf8');
const chunks=Array.from(bytes,value=>Buffer.from([value]));
const broken=chunks.map(chunk=>chunk.toString('utf8')).join('');
const decoder=new StringDecoder('utf8');
const correct=chunks.map(chunk=>decoder.write(chunk)).join('')+decoder.end();
assert.notEqual(broken,original);
assert.equal(correct,original);
assert.equal(bytes.length,12);
console.log('逐块解码损坏中文；流式解码得到团队文档；原文为 12 字节');
```

预期输出一条说明。问题不在字符串拼接本身，而在第一次解码时已经丢失了跨块字节关系。后面再把替代字符拼起来也不能恢复原文。对于按行处理的文本，还要保留最后一个未完成行片段，因为字节块同样不保证落在换行处。

## API 细读：背压、错误与完成边界

可写流的 write(chunk) 返回布尔值，不是写入成功字节数。返回 false 表示内部缓冲达到需要减速的程度，调用者应等待 drain 再继续供给；它不表示刚才那块数据被拒收。忽略这个返回值继续写，缓冲可能持续增长。highWaterMark 是开始施加背压的阈值，不是整个进程内存的硬上限，也不会阻止上游一次创建一个巨大的 Buffer。

Transform 默认按字节流工作，objectMode 则改变块的计量和语义。transform 回调接收 chunk、encoding 和 callback；异步处理后也只能调用一次 callback。忘记调用会卡住管线，调用两次会产生错误或混乱。需要并行处理文档片段时，应另设有界任务池，不要在每次 transform 中随意启动不等待的 Promise，否则输出顺序与结束时间都难以保证。

Promise 版 pipeline 会等待流链完成，成功通常兑现为 undefined；传入 signal 后可以合作取消，没有 signal 就没有这一外部取消入口。它会传播错误并处理链路销毁，但不保证所有目的地都具备事务回滚能力。文件可能已经写入一部分，远程对象也可能产生未完成分片，因此业务层仍必须在 catch 或补偿任务中清理。

`createWriteStream` 默认 flags 为 w，会创建或截断文件；教学接收核心使用 wx，以避免意外覆盖已有临时路径。返回的是流，打开文件的错误可能稍后通过 error 事件出现，不能只在创建表达式外写同步 try/catch。交给 pipeline 观察错误，比只假设 createWriteStream 调用没抛就代表文件可写更可靠。

rename 的成功表示名字变更操作完成，并不自动提供数据库事务、对象内容校验或跨文件系统复制。unlink 删除的是一个路径对应的目录项，不能拿客户端传来的任意路径交给它。清理函数应只接受服务端创建并记录的对象标识，再由受控存储层解析路径；资源所有权比在删除时临时猜测路径是否安全更重要。

## 原练习的完整参考：边写入边计算摘要

保存为 `integrity-stream.mjs`，Node 22.22 或 24 执行 `node integrity-stream.mjs`，无依赖。它对数据流同时做大小限制和 SHA-256，拒绝空内容，仅在成功后发布最终文件；最后删除自己的练习目录。摘要用于检测字节是否一致，不证明文件没有恶意内容。

完整代码已收录在本章末尾的练习参考答案中；可先阅读说明，再展开复制运行。


createHash('sha256') 创建有状态摘要对象，update 可调用多次并按顺序加入字节，digest('hex') 输出十六进制字符串并结束摘要过程。不能先 digest 再继续 update，也不能交换数据块顺序后期待同一结果。边流式更新只需要保留摘要状态，不需要在写完后 readFile 整个对象，从而避免第二次完整内存占用。

## 上传验证分成不同层次

文件名与媒体类型适合用作初步筛选，不足以证明格式。文件头特征能排除部分错误，但也不能证明整份 PDF 可以安全解析。实际处理还需要解析器验证、页面或像素限制、压缩展开限制和隔离执行环境。每层检查解决不同问题，不能拿“已校验后缀”替代内容解析，也不能把“能解析”当成可以任意公开预览。

压缩文件尤其需要关注展开后大小。十兆压缩包可能展开成远大于上传限制的数据，递归嵌套还会增加 CPU 和文件数量。应按累计展开字节、条目数量、目录深度和执行时长建立限制，并拒绝条目路径逃离目标目录。把请求体限制提高到一百兆，不能解决解析过程中的资源放大。

图片尺寸与文件字节同样不是一回事。一个压缩后不大的图片，解码成像素缓冲后可能占用大量内存。团队助手若生成缩略图，应在解码器支持的范围内尽早读取尺寸并限制像素数量，失败时保持原文档状态可解释，不要让一个缩略图异常把整个服务进程拖垮。

## 对象存储直传与服务端代理的取舍

服务端代理上传的优点是容易统一身份、大小和流式处理，代价是应用服务器承担流量与连接。浏览器直传对象存储可以减少这部分压力，但服务端仍应控制目标对象键、有效期、允许方法和必要的大小或类型条件。上传完成回调只表示客户端声称完成，后端需要核实对象存在、大小和归属，再建立 ready 状态。

不要允许客户端任意提交一个外部 URL 作为“刚上传的文件”。否则后端下载该 URL 时可能访问不应访问的内部网络，还可能拉取没有体积上限的内容。安全的直传流程通常由服务端先生成上传记录和受限目标，完成时只接受这条记录对应的对象标识。资源先有明确归属，再允许处理，能够把权限与存储路径联系起来。

大文件分片上传增加恢复能力，也增加状态复杂度。每片需要编号、大小和关联上传会话，最终合并应验证片段完整且顺序正确，并计算整体摘要。单片成功不代表整个文档可用，过期未完成会话要有清理策略。不要在没有断点续传需求时一开始就自制完整分片协议；优先评估存储服务已有的成熟能力。

## 调试与上线观察

调试文件问题应记录对象编号、字节数、阶段、耗时和公开错误码，不记录整份文件正文。检查是否在读取之前就创建了巨大 Buffer，是否在日志中序列化 Buffer，是否在 Promise.all 中同时打开过多文件。流式 API 并不能弥补业务层提前把所有文件读入数组的设计，内存观察要覆盖完整调用链。

测试至少包含中文跨块、空文件、刚好达到上限、超过一字节、来源中断、目的地不可写和取消。一次正常上传成功无法证明清理路径正确。对于生产目录，不要为了测试故障去修改已有文件权限或删除共享数据；应在专属练习目录构造失败，验证只清理自己的对象。

最终发布可以是数据库版本指针切换、对象标签变更或索引版本更新。重要的是消费者只读取已确认完成的版本，失败产物不会冒充最新内容。旧版本保留多久、用户删除后如何清理派生文件、备份是否仍保留内容，都应成为生命周期设计的一部分。文件系统 API 解决字节操作，业务服务负责这些更长的承诺。

## 路径检查与实际操作之间也有竞争

先检查文件不存在，再用普通写入模式创建，并不能保证中间没有另一个请求创建同名文件。应该让创建动作本身具备排他语义，并处理已存在错误。类似地，先检查目录权限再写入，只能提供一瞬间的观察，实际打开仍可能失败。文件系统操作的错误不是多余保险，而是并发世界中最终的判断依据。

把路径规范化并确认位于目标目录之下，可以阻止很多明显的路径穿越，但受控目录中的符号链接或 Windows 重解析点还可能改变实际目标。因此上传存储目录应由服务端管理，不允许普通用户随意创建链接或目录结构。业务层只传对象编号，存储层统一映射路径，比每个路由各自拼一套目录更容易维护安全边界。

磁盘满、配额不足和权限拒绝需要保留具体内部错误，向用户则返回稳定的上传失败状态。不能在清理失败时直接返回上传成功，也不能无限重试同一个不可写磁盘。清理器应记录失败对象和重试次数，让运维能看见积压的临时文件；否则看似已经在 catch 中处理的异常，最终会以磁盘逐渐耗尽的方式再次出现。


## 练习：计算摘要并拒绝空文件

要求在同一条流里计算 SHA-256，返回十六进制摘要；零字节输入必须失败并清理临时文件。提示：在 Transform 中更新哈希，在 pipeline 成功后检查 bytes，再决定是否 rename。不要为了计算摘要再 readFile 把完整文件读回内存。

<details><summary>参考答案（含完整可运行实现）</summary>

从 node:crypto 导入 createHash，进入 saveStream 后创建哈希对象，每次通过大小检查的块调用 hash.update(chunk)。pipeline 结束后如果 bytes 等于零就抛出 EMPTY_FILE，原 catch 会删除临时文件；否则完成 rename 后调用 hash.digest('hex') 并加入返回值。新增断言比较 Buffer.from('知识库') 的离线摘要，并用 Readable.from([]) 验证空文件拒绝。下面给出包含这些行为的完整参考文件。





```js integrity-stream.mjs
// integrity-stream.mjs
import { Readable,Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createWriteStream } from 'node:fs';
import { mkdtemp,rename,unlink,readdir,rmdir } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { createHash,randomUUID } from 'node:crypto';
import assert from 'node:assert/strict';
async function save(source,directory,maxBytes=1024) {
  if(!Number.isSafeInteger(maxBytes)||maxBytes<1) throw new RangeError('大小上限无效');
  const id=randomUUID();const partial=join(directory,`${id}.part`);const final=join(directory,`${id}.bin`);
  const hash=createHash('sha256');let bytes=0;
  const guard=new Transform({transform(chunk,encoding,callback){
    bytes+=chunk.length;
    if(bytes>maxBytes){callback(Object.assign(new Error('内容超限'),{code:'TOO_LARGE'}));return;}
    hash.update(chunk);callback(null,chunk);
  }});
  try {
    await pipeline(source,guard,createWriteStream(partial,{flags:'wx',mode:0o600}));
    if(bytes===0) throw Object.assign(new Error('不接受空文件'),{code:'EMPTY_FILE'});
    const digest=hash.digest('hex');
    await rename(partial,final);
    return {path:final,bytes,digest};
  } catch(error) {
    try {await unlink(partial);} catch(cleanup){if(cleanup.code!=='ENOENT')throw new AggregateError([error,cleanup],'处理与清理都失败');}
    throw error;
  }
}
const directory=await mkdtemp(join(tmpdir(),'integrity-lesson-'));
try {
  const data=Buffer.from('团队文档');
  const result=await save(Readable.from([data.subarray(0,2),data.subarray(2)]),directory);
  assert.equal(result.bytes,12);
  assert.equal(result.digest,createHash('sha256').update(data).digest('hex'));
  await assert.rejects(()=>save(Readable.from([]),directory),{code:'EMPTY_FILE'});
  await assert.rejects(()=>save(Readable.from([Buffer.alloc(20)]),directory,10),{code:'TOO_LARGE'});
  const names=await readdir(directory);
  assert.equal(names.length,1);assert.ok(!names[0].endsWith('.part'));
  console.log('摘要正确；空文件与超限被拒绝；仅保留一个成功文件');
} finally {
  for(const name of await readdir(directory))await unlink(join(directory,name));
  await rmdir(directory);
}
```


</details>

## 可验证的验收标准

正常文字按 UTF-8 字节计数为九；超限失败后没有 .part；同一输入的摘要稳定；零字节策略有明确结果；取消时尽力清理临时文件。能解释 pipeline 只负责流的资源传播，不负责删除数据库记录，也不保证 HTTP 错误响应一定已经送达。

## 三个自测问题与答案

1. 为什么不能以 File.name 直接生成磁盘路径？答案：它来自不可信客户端，可能包含路径穿越、冲突与平台特殊名称。
2. 背压解决文件总大小限制吗？答案：不解决，背压控制缓冲与速度，大小限制仍需显式累计字节。
3. 上传成功是否代表文档可被模型检索？答案：不代表，还需要校验、解析、切分、索引和状态发布，各阶段都可能失败。


## 本章验证记录

编写时使用 Node 22.22.0 对本章全部 3 个 JavaScript 完整文件执行了语法检查。已在本机实际运行通过：`save-stream.mjs`、`utf8-stream.mjs`、`integrity-stream.mjs`。

## 本章官方参考

- [Node Streams](https://nodejs.org/api/stream.html)：背压、Transform 和 pipeline。
- [Node File System](https://nodejs.org/api/fs.html)：排他创建、重命名与文件删除。
- [OWASP File Upload](https://cheatsheetseries.owasp.org/cheatsheets/File_Upload_Cheat_Sheet.html)：类型、存储位置与资源限制。
- [Node String decoder](https://nodejs.org/api/string_decoder.html)：跨字节块的文本解码。