批量导入 1000 条对话：成本拆分与性能验收

批量导入的成本来自多个阶段：发现来源、读取和解析、归一化、SQL 写入、数据库导出；摘要、嵌入和关系发现又是另一条链路。先分开计时，才能知道该减少解析对象、导出次数，还是模型请求。本文保留“1000 条”作为合成样本规模，不把原稿的个人机器耗时、OOM 和优化倍数当作已验证结果。

固定实现已经做了什么

server/src/services/import.ts 先对启用的适配器依次 detect/scan，把全部元数据收集到 allMetas，然后逐条检查、解析和写入。大小与修改时间相同可在解析前跳过；元数据变化但内容哈希相同只更新元数据。它不会因为已经记录 parserVersion 就强制重解析，见适配器内部链路。

每篇解析结果仍是完整 messages 数组。写入用单篇事务和预编译语句插入消息；替换会失效旧的自动笔记并安排向量清理，同时保留符合条件的用户编辑笔记或拒绝状态。批次结束调用 saveDatabase。server/src/db/index.ts 的自动保存仍由服务启动，没有在 importAll 中临时关闭；“每条都导出、导入时自动停保存、已经流式 parseAndInsert”都不是这份提交的行为。

导入返回 imported、replaced、skipped、errors 以及具体 ID 和 summarizationCandidateIds。候选 ID 不是已完成摘要。server/src/services/summarize.ts 才会执行经验筛选、摘要、嵌入和关系发现；摘要可被 filtered，嵌入失败也不一定使摘要失败。

给各阶段定义可核对的成本

阶段	同时记录	先验证的条件
detect/scan	来源、文件数、元数据数、耗时	可见文件与预期来源一致
parse/hash	消息数、输入字节、最大单篇大小	内容字段和顺序保持一致
SQL/事务	新增、替换、跳过、失败 ID	失败篇不留半条会话
export/save	导出字节、调用次数、重开结果	内存提交与文件落盘分开验证
摘要/嵌入/关系	入队与最终状态、请求数、错误	队列收据不等于完成或质量

Node 的 memoryUsage 文档区分 RSS、V8 heap 和 external；arrayBuffers 已包含在 external 中，不要重复相加。定时采样可能漏掉同步 export 的短峰值，末尾快照更不是整次峰值。记录采样方法、Node/依赖版本、硬件、样本和冷暖缓存条件；不要用对话数量推导固定内存门槛。

用 1000 条合成输入比较保留范围

在独立目录用 Node 24 保存下面文件为 performance.mjs，先运行 npm install --ignore-scripts --save-exact sql.js@1.14.2，再运行 node performance.mjs。程序只创建临时归一化 JSONL 和独立 WASM 数据库：一条路径保留全部输入对象，另一条用 readline 逐行插入；两者都是每篇事务、每篇预编译消息语句、最后一次导出并重开检查。

它比较教学输入的对象保留范围和最终数据，不运行 ChatCrystal 的 detect/scan、厂商解析器、自动保存或模型；更不表示当前适配器已经支持边读边插入。计数中的 1000 与 1 是显式保留的归一化对象数，不是缓冲区大小、整个数据库内存或进程峰值。计时及内存快照会随机器和运行顺序变化。

import assert from 'node:assert/strict';
import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises';
import { createReadStream } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { createInterface } from 'node:readline';
import { performance } from 'node:perf_hooks';
import initSqlJs from 'sql.js';

const SQL = await initSqlJs();
const dir = await mkdtemp(join(tmpdir(), 'import-fixture-'));
const file = join(dir, 'normalized.jsonl');
const count = 1000;
const rows = Array.from({ length: count }, (_, i) => ({
  id: `fixture-${i}`,
  messages: [
    { role: 'user', content: `question ${i}` },
    { role: 'assistant', content: `answer ${i}` },
  ],
}));
await writeFile(file, rows.map(JSON.stringify).join('\n') + '\n');
rows.length = 0;

async function run(retainAll) {
  const db = new SQL.Database();
  db.run(
    'CREATE TABLE messages (conversation TEXT, position INTEGER, role TEXT, content TEXT, PRIMARY KEY (conversation, position))'
  );
  let peakRetained = 0;
  let imported = 0;
  const started = performance.now();
  const insert = (item) => {
    assert.equal(item.messages.length, 2);
    db.run('BEGIN');
    const stmt = db.prepare('INSERT INTO messages VALUES (?, ?, ?, ?)');
    try {
      item.messages.forEach((m, i) => stmt.run([item.id, i, m.role, m.content]));
      db.run('COMMIT');
      imported++;
    } catch (error) {
      db.run('ROLLBACK');
      throw error;
    } finally {
      stmt.free();
    }
  };
  try {
    if (retainAll) {
      const items = (await readFile(file, 'utf8')).trim().split('\n').map(JSON.parse);
      peakRetained = items.length;
      items.forEach(insert);
    } else {
      const lines = createInterface({ input: createReadStream(file), crlfDelay: Infinity });
      try {
        for await (const line of lines) {
          if (!line.trim()) continue;
          peakRetained = Math.max(peakRetained, 1);
          insert(JSON.parse(line));
        }
      } finally {
        lines.close();
      }
    }
    const values = db.exec(
      'SELECT conversation, position, role, content FROM messages ORDER BY conversation, position'
    )[0].values;
    assert.equal(imported, count);
    assert.equal(values.length, count * 2);
    const bytes = db.export();
    const reopened = new SQL.Database(bytes);
    try {
      assert.equal(reopened.exec('SELECT count(*) FROM messages')[0].values[0][0], count * 2);
    } finally {
      reopened.close();
    }
    return {
      values,
      peakRetained,
      elapsedMs: performance.now() - started,
      exportedBytes: bytes.length,
      memory: process.memoryUsage(),
    };
  } finally {
    db.close();
  }
}

try {
  const retained = await run(true);
  const streamed = await run(false);
  assert.deepEqual(streamed.values, retained.values);
  assert.equal(retained.peakRetained, count);
  assert.equal(streamed.peakRetained, 1);
  const { values: a, ...first } = retained;
  const { values: b, ...second } = streamed;
  console.log(
    JSON.stringify(
      {
        scope: 'synthetic normalized JSONL; not ChatCrystal importAll',
        retained: first,
        streamed: second,
      },
      null,
      2
    )
  );
} finally {
  await rm(dir, { recursive: true, force: true });
}

优化前先让验收失败一次

将逐行路径的 peakRetained 改成 count，保留范围断言应失败；只插入 user 消息，消息数量和等价性验收应失败。恢复代码再运行。程序输出 elapsedMs 和末尾 memory，仅作本次观察；两条路径在同一进程顺序执行，不能据此宣布速度或 RSS 获胜。

对真实导入另做故障样本：坏 JSON、超长单篇、重复/改写内容、不可写输出和落盘重开。流式读取减少某类输入对象，不会让 WASM 数据库变成外存数据库。先保证事务、内容哈希、替换语义和可恢复性，再决定是否值得增加异步流式接口，参见WASM 持久化边界。

摘要队列和监听器分别验收

server/src/queue/index.ts 的并发为 1，intervalCap 为 1、interval 为 1000ms；这是队列任务启动节奏，不是所有 Provider HTTP 请求的统一限速。内部 SDK 重试和任务重试可能增加请求；外层只在错误字符串包含 429 且仍有机会时指数等待，默认最多四次任务尝试。不能用入队数估算真实请求数或完成时间。

server/src/watcher/index.ts 在文件稳定等待之外再防抖 3 秒，同一 watcher 内已有导入时置 pending，结束后延迟重跑。该布尔量属于 watcher，不是所有 HTTP/CLI 导入共享的全局锁；云模式由 watcherPolicy 禁止监听。批量测试应固定入口并观察任务状态，避免把文件事件、解析次数和模型请求混成一个吞吐数字。