从 5.4 秒到 106 毫秒：一个 EventV2 批量写入问题的定位与修复

背景

KiloCode 的 fork 功能可以从一个会话的某个时间点分出一个新会话，继承之前所有消息。和 Git 的分支类似，但操作对象是 LLM 会话，每次 fork 需要复制源会话中的全部消息。

我比较依赖这个功能，但涉及数十万词元的会话，fork 的耗时让人不太舒服。一个中等规模的测试会话——385 条消息、1430 个片段、约 20 万 token——fork 耗时约 5 秒。用户侧可以看到新会话正在创建，但消息是一条一条出现的。与其每次等，不如翻源码看看慢在哪里。

原始实现

fork 的核心实现是一个逐条复制的循环：

for (const msg of msgs) {
  const cloned = yield* updateMessage({...msg.info})
  for (const part of msg.parts) {
    yield* updatePart(part)
  }
}

updateMessage 和 updatePart 分别负责写入单条消息或片段。每个函数内部调用了 events.publish()——EventV2 框架的入口，用于发布一个持久化事件。

EventV2 发布链路

events.publish() 的完整调用链如下：

events.publish(SessionV1.Event.MessageUpdated, data)
  → commitSyncEvent()
    → SQLite BEGIN IMMEDIATE
      → beforeCommit() guard 运行
      → projectors 运行
        INSERT INTO MessageTable / PartTable
        ON CONFLICT DO UPDATE
      → EventSequenceTable 更新（seq = seq + 1）
      → EventTable 插入（id, aggregate_id, seq, type, data）
    → SQLite COMMIT
  → sync handlers 运行（SSE 广播）
  → listeners 运行

关键点在于：事件发布和业务数据写入在同一个 SQLite 事务中。数据由 projector 在事务内写入 MessageTable 和 PartTable，EventV2 框架再写入 EventTable 和 EventSequenceTable。对这个会话而言，385 条消息 × 1430 个片段 = 1815 次 events.publish()，每次独立事务。

定位瓶颈

将 fork 过程拆为三个阶段并分别计时：

1. collect：读取源会话数据到内存
2. db：写入 MessageTable 和 PartTable
3. event：通过 EventV2 逐条发布事件

计时写到一个临时文件中，而非标准输出（TUI 进程会重定向标准输出）。结果：

fork: msgs=385 parts=1430 collect=22ms db=370ms event=5048ms total=5440ms

event 阶段 5048ms，占总耗时 92.8%。数据库写入仅 370ms。

方向一：优化数据库写入（无效）

第一个思路是批量化数据库写入，把循环内逐条的 updateMessage 和 updatePart 拆成两个阶段：先在内存中收集数据，再用单次事务批量写入 MessageTable 和 PartTable。后续 events.publish() 调用保持原样。

// 收集阶段——零数据库操作
for (const msg of msgs) { batchMessages.push(clonedMessage); batchParts.push(...); eventsToEmit.push(...) }

// 写入阶段——单次事务批量写入
yield* db.transaction(tx => {
  tx.insert(MessageTable).values(batchMessages)
  tx.insert(PartTable).values(batchParts)
})

// 发布阶段——逐条发布事件
for (const emit of eventsToEmit) { yield* emit() }

结果没有改善：

fork: msgs=394 parts=1485 collect=31ms db=472ms event=6020ms total=6523ms

event 阶段反而增加了。原因在于：数据库写入的批量化没有减少事件发布的次数。1815 次 events.publish() 仍然各自触发一次 SQLite 事务（写入 EventTable + EventSequenceTable）、一次 EventWire.encode 序列化、一次 GlobalBus.emit SSE 广播。瓶颈在于循环次数本身，而不在于循环中某个子操作的性能。

方向二：事务嵌套（无效）

假设问题在于每次 commitSyncEvent 独立打开一个 SQLite 事务，考虑用外层 db.transaction() 包裹整个循环，使内部事务退化为 SQLite savepoint 以减少事务开销：

yield* db.transaction((tx) =>
  Effect.gen(function* () {
    for (const emit of eventsToEmit) { yield* emit() }
  }),
)

event 阶段从 5048ms 升到 6020ms。savepoint 比独立事务轻量，但 EventWire.encode 序列化、GlobalBus.emit SSE 广播、projector 的空更新执行等操作并没有减少。在这个场景中，问题不在事务类型而在循环次数。

方向三：跳过 EventV2 逐条事件

如果循环根本上就不应该存在，那么能否跳过这些逐条事件？

为什么可以跳过

events.publish(SessionV1.Event.MessageUpdated, ...) 和 events.publish(SessionV1.Event.PartUpdated, ...) 在 EventV2 框架中同时做三件事：

1. 写入 MessageTable / PartTable（通过 projector 在 commitSyncEvent 事务内执行）
2. 写入 EventTable / EventSequenceTable（事件溯源记录）
3. SSE 广播到 TUI 客户端（通过 GlobalBus）

对 fork 场景而言：（1）的业务数据写入可以直接由我们自己的 db.transaction() 完成，不需要 projector 介入。（2）的事件溯源记录在 fork 场景下没有实际用途——新会话的数据已经完整存在于业务表中，TUI 不需要回放事件来重建会话状态。（3）的 SSE 广播是为了让 TUI 实时显示消息，但 fork 操作完成后 TUI 会导航到新会话，导航时会触发 sync.sync(sessionID) 从数据库批量加载所有消息。

没有其他系统组件监听 MessageUpdated 和 PartUpdated 事件。排除了 plugin hooks、ACL（agent control layer）和其他事件订阅者后，可以确认跳过这两个事件类型在当前架构下是安全的。

ForkCompleted 信号

需要一个通知机制替代逐条事件的作用：告诉 TUI「fork 已经完成，可以加载数据」。新增一个事件类型：

export const ForkCompleted = EventV2.define({
  type: "session.fork.completed",
  schema: { sessionID: SessionSchema.ID, sourceSessionID: SessionSchema.ID },
})

该事件只携带源会话和目标会话的 ID，不携带业务数据。TUI 侧收到后调用 sync.sync(sessionID)，由该函数通过 /messages API 一次性加载全部消息。

重构后的 fork 流程

// 收集阶段
const batchMessages: typeof MessageTable.$inferInsert[] = []
const batchParts: typeof PartTable.$inferInsert[] = []
for (const msg of msgs) {
  batchMessages.push(clonedMessage); batchParts.push(...processedParts)
}

// 单次事务写入业务表
yield* db.transaction((tx) => {
  tx.insert(MessageTable).values(batchMessages).onConflictDoNothing()
  tx.insert(PartTable).values(batchParts).onConflictDoNothing()
})

// 单个 ForkCompleted 信号
yield* events.publish(ForkCompleted, { sessionID: session.id, sourceSessionID: input.sessionID })

结果：

fork: msgs=433 parts=1639 collect=32ms db=426ms event=0ms total=458ms

耗时从 5440ms 降到 458ms（12 倍）。事件发布阶段归零，因为原本 1815 次事件被替换为 1 个 ForkCompleted。

多行 VALUES INSERT

458ms 中数据库写入仍占 426ms。写入部分的实现仍是逐行 INSERT：

for (const m of batchMessages) tx.insert(MessageTable).values(m).onConflictDoNothing()
for (const p of batchParts) tx.insert(PartTable).values(p).onConflictDoNothing()

drizzle-orm 的 values() 方法接受数组，会展开为 SQLite 的多行 VALUES 语法。改为：

tx.insert(MessageTable).values(batchMessages).onConflictDoNothing()
tx.insert(PartTable).values(batchParts).onConflictDoNothing()

翻译后的 SQL 为：

INSERT OR IGNORE INTO message (id, session_id, time_created, data) VALUES (row1), (row2), ...
INSERT OR IGNORE INTO part (id, message_id, session_id, time_created, data) VALUES (row1), (row2), ...

SQLite 解析一次 SQL 语法树、加一次写锁、写一次 WAL 日志即可完成所有数据写入。逐行 INSERT 则需要解析 2072 次、加锁 2072 次、WAL 写入 2072 次。分开计时：

fork_detail: msgs=330 parts=1226 db_msg=12ms db_part=47ms
total: db_transaction=80ms overall=106ms

数据库写入从 426ms 降到 80ms。

耗时记录

总结

这个优化做了三次调整。前两次尝试优化循环内部的子操作，但问题在于循环次数本身——1815 次迭代中，每次迭代的固定开销是 2.8ms，不管怎样分摊事务、聚合写入，只要循环还在，总时间就由 1815 × 2.8ms 决定。第三次调整不再围绕循环做文章，而是直接问这个循环是否需要存在。

updateMessage / updatePart 被设计用于流式输出场景：LLM 生成一条消息时，工具调用、文本片段、推理过程需要分多次写入和广播。fork 操作是一次性复制大量数据，粒度要求完全不同。同一套接口被两个粒度不同的场景共用，是性能问题的根因。

更一般地看，事件溯源框架的写入接口天然面向逐条增量写入：每条事件独立持久化、独立广播。这个设计在流式场景下合理且必要。但当数据量大、且不需要中间状态时，逐条写入的固定开销会累积成显著延迟。框架层面是否应该提供批量写入通道、是否应该区分「需要实时广播的事件」和「只需持久化的事件」，这组取舍在不同场景下会有不同的答案。

多行 VALUES 的改动只涉及 2 行代码，将数据库写入从 426ms 降到 80ms。这个效果不是优化了某条 SQL 的执行计划，而是改变了 SQL 语句本身的形式——从 2072 条独立 INSERT 变成 2 条多行 INSERT，SQLite 的解析次数和锁定次数都从 2072 降到了 2。每个操作本身的开销不大，关键在于它被重复执行了多少次。这和主问题的根因其实是一样的。