fix(session-persistence-jsonl): drop crash-unsafe cross-process log lock

The product model has no cross-process writer exclusion (the coordinator serializes per-session operations in-process; the README documents one live writer per session), so the wx-created .lock sibling only guarded byte-level races while adding two failure modes: a crash leaves a stale lock that permanently wedges that log's appends/repairs/replacements, and a post-commit lock cleanup failure makes a committed append look failed, so the retained write-behind batch retries into duplicate seqs. Remove withLogLock and keep replaceStored's revision compare-and-swap at the commit boundary (recheck immediately before the atomic rename).
2026-08-19 20:42:56 +08:00
1 parent e8f4315cee
commit 270a06b38b
@@ -22,7 +22,7 @@ Session logs must be upgradable after release, and the runtime that ships first

 **Recovery and writeback consume current-format data.** `inspect()` and `readFrom()` decode only in memory. Cold `prepare()`/`load()` first decode the whole source, add the current recovery closers, and replace the exact old revision with that complete balanced current-format stream. Live HMR adoption uses the same replacement primitive after seed verification but does not synthesize closers for a turn still owned by the live Session. A successful replacement or revision conflict discards the prepared object and reopens the stored source before continuing.

-**Replacement is an internal backend compare-and-swap.** `replaceStored(expectedRevision, meta, events)` accepts a streaming current-format log and checks storage identity plus the source revision inside the commit exclusion. JSONL writes and fsyncs a sibling temporary artifact, rechecks the revision while holding its cross-process lock, atomically replaces the path (using the Windows write-through replacement primitive there), and syncs the parent directory on POSIX. SQLite stages the event iterator, then rechecks and replaces the header and event rows in one transaction. A failed commit leaves one complete old or new log; retaining a permanent pre-upgrade copy is a separate recovery policy, not part of the format migration API.
+**Replacement is an internal backend compare-and-swap.** `replaceStored(expectedRevision, meta, events)` accepts a streaming current-format log and checks storage identity plus the source revision at the commit boundary. JSONL writes and fsyncs a sibling temporary artifact, rechecks the source revision immediately before the atomic replace, atomically replaces the path (using the Windows write-through replacement primitive there), and syncs the parent directory on POSIX; like every other coordinator freshness check, the recheck adds no cross-process writer exclusion — JSONL assumes one live writer per session. SQLite stages the event iterator, then rechecks and replaces the header and event rows in one transaction. A failed commit leaves one complete old or new log; retaining a permanent pre-upgrade copy is a separate recovery policy, not part of the format migration API.

 **A per-event `ignorable` marker covers vocabulary growth, so ordinary event additions never bump the version.** The event vocabulary is decided by which plugins are mounted, which a single version integer cannot describe. A reader meeting an unrecognized event type refuses to interpret the log unless the event carries `ignorable: true` in its envelope. The default is *required*: forgetting the marker over-refuses a resumable session (an inconvenience), while a default of ignorable would make the same mistake silently resume a gutted one (a safety failure). The architecture makes this sound: model-visible content flows only through the three `surfaceOp`-marked surface event types plus the `request/header`/`request/context` folds, so the dangerous unknowns are exactly the non-surface events that change how the rest of the log is read (`session/end-seed` is the existing example).

@@ -22,7 +22,7 @@ Session log 在发布后必须能升级格式，而最先发布的运行时决

 **Recovery 和写回只消费当前格式数据。**`inspect()` 和 `readFrom()` 只在内存中解码。Cold `prepare()`/`load()` 先解码完整 source，补充当前 recovery closers，再用完整、平衡的当前格式 stream 替换精确的旧 revision。Live HMR adoption 在 seed 校验后使用同一个 replacement primitive，但不会为仍由 live Session 掌握的 turn 合成 closer。替换成功或 revision 冲突后都会丢弃 prepared object，重新打开持久化 source 后再继续。

-**Replacement 是 backend 内部的 compare-and-swap。**`replaceStored(expectedRevision, meta, events)` 接受流式当前格式日志，并在提交排他区内检查存储身份和 source revision。JSONL 写入并 fsync 同目录临时 artifact，在持有跨进程锁时复核 revision，原子替换路径（Windows 使用 write-through replacement primitive），并在 POSIX 上同步父目录。SQLite 先暂存 event iterator，再在一个事务中复核并替换 header 与 event rows。提交失败后只会留下完整旧日志或完整新日志；永久保留升级前副本是独立的恢复策略，不属于 format migration API。
+**Replacement 是 backend 内部的 compare-and-swap。**`replaceStored(expectedRevision, meta, events)` 接受流式当前格式日志，并在提交边界检查存储身份和 source revision。JSONL 写入并 fsync 同目录临时 artifact，在原子替换路径前立即复核 source revision，然后原子替换（Windows 使用 write-through replacement primitive），并在 POSIX 上同步父目录；与协调器的其他新鲜性检查一样，复核不提供跨进程写者排他——JSONL 假定每个 session 同时只有一个 live writer。SQLite 先暂存 event iterator，再在一个事务中复核并替换 header 与 event rows。提交失败后只会留下完整旧日志或完整新日志；永久保留升级前副本是独立的恢复策略，不属于 format migration API。

 **逐事件的 `ignorable` 标记吸收词汇表增长，普通的新增事件永远不用升版本。**事件词汇表由挂载了哪些插件决定，单个版本整数描述不了它。读取器遇到不认识的事件类型时拒绝解读日志，除非该事件的信封带 `ignorable: true`。默认为必需：忘写标记的后果是把一个本可恢复的会话拒绝过头（体验问题），而默认可忽略会让同样的疏忽静默恢复出残缺会话（安全事故）。架构保证了这条规则成立：模型可见内容只经三种带 `surfaceOp` 标记的 surface 事件加 `request/header`、`request/context` 折叠进入重建，危险的未知事件恰好是那些不进 surface 但改变日志其余部分解读方式的事件（`session/end-seed` 是现存例子）。

@@ -9,7 +9,7 @@
 import { Context } from '@deepseek-ai/cordis'
 import z from '@deepseek-ai/schemastery'
 import { readdirSync } from 'node:fs'
-import { open, mkdir, readFile, readdir, realpath, link, rename, rm, stat, truncate, writeFile } from 'node:fs/promises'
+import { open, mkdir, readFile, readdir, realpath, link, rename, rm, stat, truncate } from 'node:fs/promises'
 import { dirname, join, resolve } from 'node:path'
 import { performance } from 'node:perf_hooks'
 import { scheduler } from 'node:timers/promises'
@@ -44,9 +44,6 @@ const DEFAULT_COMPRESSION: JsonlCompression = 'zstd'
 * remains an indivisible synchronous decode.
 */
 const ZSTD_DECODE_YIELD_INTERVAL_MS = 500
-const LOG_LOCK_RETRY_INITIAL_MS = 20
-const LOG_LOCK_RETRY_MAX_MS = 200
-const LOG_LOCK_TIMEOUT_MS = 2_000
 const REPLACEMENT_BATCH_SIZE = 128

 /** Assert that the independently decodable first frame contains only the header record. */
@@ -130,10 +127,6 @@ function isENOENT(error: unknown): boolean {
  return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
 }

-function isEEXIST(error: unknown): boolean {
-  return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST'
-}
-
 /**
 * The JSONL persistence backend. Load as a plugin; it registers as
 * `ctx.sessionPersistence` and (via the coordinator) installs the write-path
@@ -484,8 +477,7 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
  async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
    await this.ensureRootEncoding()
    if (isMaterialized) {
-      const path = logPath(this.root, meta.cwd, meta.id, this.compression)
-      await this.withLogLock(path, () => this.appendLines(meta, events))
+      await this.appendLines(meta, events)
    } else {
      await this.materialize(meta, events)
    }
@@ -501,12 +493,9 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
    tornMarker: JsonlTornMarker | undefined,
    closers: readonly SessionEvent[],
  ): Promise<void> {
-    const path = logPath(this.root, meta.cwd, meta.id, this.compression)
-    await this.withLogLock(path, async () => {
-      if (tornMarker !== undefined) await this.repair(meta, tornMarker.truncateTo)
-      const repairedEvents = [...(tornMarker?.recoveredEvents ?? []), ...closers]
-      if (repairedEvents.length > 0) await this.appendLines(meta, repairedEvents)
-    })
+    if (tornMarker !== undefined) await this.repair(meta, tornMarker.truncateTo)
+    const repairedEvents = [...(tornMarker?.recoveredEvents ?? []), ...closers]
+    if (repairedEvents.length > 0) await this.appendLines(meta, repairedEvents)
  }

  /** Replace one exact source revision through a synced sibling and atomic namespace update. */
@@ -522,48 +511,46 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
        `session "${meta.id}" no longer has revision ${expectedRevision}`,
      )
    }
-    await this.withLogLock(path, async () => {
-      let current: JsonlStoredHeader
-      try {
-        current = await this.readStoredHeader(path, meta.id)
-      } catch (error: unknown) {
-        if (isENOENT(error)) {
-          throw new SessionPersistenceRevisionConflictError(
-            `session "${meta.id}" no longer has revision ${expectedRevision}`,
-          )
-        }
-        throw error
+    let current: JsonlStoredHeader
+    try {
+      current = await this.readStoredHeader(path, meta.id)
+    } catch (error: unknown) {
+      if (isENOENT(error)) {
+        throw new SessionPersistenceRevisionConflictError(
+          `session "${meta.id}" no longer has revision ${expectedRevision}`,
+        )
      }
-      if (current.revision !== expectedRevision) {
+      throw error
+    }
+    if (current.revision !== expectedRevision) {
+      throw new SessionPersistenceRevisionConflictError(
+        `session "${meta.id}" changed before replacement of revision ${expectedRevision}`,
+      )
+    }
+    const currentIdentity = this.storedIdentity(current.meta, path)
+    if (meta.cwd !== currentIdentity.cwd) {
+      throw new Error(`replacement for session "${meta.id}" changes its stored identity`)
+    }
+
+    const tmp = `${path}.${randomBytes(6).toString('hex')}.upgrade.tmp`
+    try {
+      await this.writeReplacement(tmp, meta, events)
+      const beforeCommit = fileRevision(await stat(path, { bigint: true }))
+      if (beforeCommit !== expectedRevision) {
        throw new SessionPersistenceRevisionConflictError(
          `session "${meta.id}" changed before replacement of revision ${expectedRevision}`,
        )
      }
-      const currentIdentity = this.storedIdentity(current.meta, path)
-      if (meta.cwd !== currentIdentity.cwd) {
-        throw new Error(`replacement for session "${meta.id}" changes its stored identity`)
+      /* v8 ignore next -- Windows uses its write-through replacement primitive. */
+      if (process.platform === 'win32') {
+        await replaceFileWin32(tmp, path)
+      } else {
+        await rename(tmp, path)
+        await this.syncDirPosix(dirname(path))
      }
-
-      const tmp = `${path}.${randomBytes(6).toString('hex')}.upgrade.tmp`
-      try {
-        await this.writeReplacement(tmp, meta, events)
-        const beforeCommit = fileRevision(await stat(path, { bigint: true }))
-        if (beforeCommit !== expectedRevision) {
-          throw new SessionPersistenceRevisionConflictError(
-            `session "${meta.id}" changed before replacement of revision ${expectedRevision}`,
-          )
-        }
-        /* v8 ignore next -- Windows uses its write-through replacement primitive. */
-        if (process.platform === 'win32') {
-          await replaceFileWin32(tmp, path)
-        } else {
-          await rename(tmp, path)
-          await this.syncDirPosix(dirname(path))
-        }
-      } finally {
-        await rm(tmp, { force: true })
-      }
-    })
+    } finally {
+      await rm(tmp, { force: true })
+    }
  }

  /** List valid unique stored sessions' metadata (header line only — no full-log parse). */
@@ -768,31 +755,6 @@ export class JsonlSessionPersistence extends SessionPersistence implements Persi
    }
  }

-  /** Serialize cooperating cross-process mutations of one materialized log. */
-  private async withLogLock<T>(path: string, operation: () => Promise<T>): Promise<T> {
-    const lockPath = `${path}.lock`
-    const deadline = Date.now() + LOG_LOCK_TIMEOUT_MS
-    let delay = LOG_LOCK_RETRY_INITIAL_MS
-    for (;;) {
-      try {
-        await writeFile(lockPath, `${process.pid}\n`, { mode: 0o600, flag: 'wx' })
-        break
-      } catch (error: unknown) {
-        if (!isEEXIST(error)) throw error
-      }
-      if (Date.now() >= deadline) {
-        throw new Error(`session log writer lock timed out at "${lockPath}"`)
-      }
-      await new Promise(resolve => setTimeout(resolve, delay))
-      delay = Math.min(delay * 2, LOG_LOCK_RETRY_MAX_MS)
-    }
-    try {
-      return await operation()
-    } finally {
-      await rm(lockPath, { force: true })
-    }
-  }
-
  /** Encode the header and first batch without combining their frame boundaries. */
  private async encodeMaterialization(meta: SessionHeader, events: readonly SessionEvent[]): Promise<Buffer | string> {
    const header = JSON.stringify(toHeaderLine(meta)) + '\n'
@@ -672,32 +672,6 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
    expect(await collectStoredRead(replaced.readEvents())).toHaveLength(128)
  })

-  it('retries a held writer lock and rejects lock acquisition failures and timeout', async () => {
-    const persistence = ctx.sessionPersistence as JsonlSessionPersistence
-    const internals = persistence as unknown as {
-      withLogLock<T>(path: string, operation: () => Promise<T>): Promise<T>
-    }
-    const path = join(root, 'lock-target')
-    const lockPath = `${path}.lock`
-    await writeFile(lockPath, 'held\n')
-    const released = new Promise<void>((resolveRelease) => {
-      setTimeout(() => { void rm(lockPath).then(() => { resolveRelease() }) }, 5)
-    })
-    await expect(internals.withLogLock(path, async () => 'committed')).resolves.toBe('committed')
-    await released
-
-    await expect(internals.withLogLock(`${root}\0invalid`, async () => undefined))
-      .rejects.toMatchObject({ code: 'ERR_INVALID_ARG_VALUE' })
-
-    await writeFile(lockPath, 'held\n')
-    const now = vi.spyOn(Date, 'now')
-      .mockReturnValueOnce(1_000)
-      .mockReturnValueOnce(3_001)
-    await expect(internals.withLogLock(path, async () => undefined))
-      .rejects.toThrow(/writer lock timed out/)
-    now.mockRestore()
-  })
-
  it('rejects malformed version-independent storage identity fields', async () => {
    const persistence = ctx.sessionPersistence as JsonlSessionPersistence
    const path = rawLogPath(root, '/work', SessionId('identity-fields'))
@@ -58,7 +58,7 @@ The `PersistenceBackend<TornMarker>` hooks (the only contract between the coordi
 | `readStoredRevision(id, signal?)` | Read the current source-qualified revision for one id without loading its event log. It uses the same revision representation as `openStored` and returns `undefined` when the id is absent. |
 | `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. |
 | `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). |
-| `replaceStored(expectedRevision, meta, events)` | Atomically replace one exact revision with a complete current-format header and event stream. Revision and stored identity checks occur inside the same backend exclusion or transaction as the commit; a mismatch rejects with `SessionPersistenceRevisionConflictError`. |
+| `replaceStored(expectedRevision, meta, events)` | Atomically replace one exact revision with a complete current-format header and event stream. Revision and stored identity checks occur at the commit boundary — immediately before the atomic rename on JSONL, inside the replacing transaction on SQLite; the checks add no cross-process writer exclusion. A mismatch rejects with `SessionPersistenceRevisionConflictError`. |
 | `list(signal?)` | List all stored metadata, observing optional cancellation. |
 | `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. |

@@ -58,7 +58,7 @@ v0 decoder 还识别[消息标识机制引入前的消息](../../../.agents/note
 | `readStoredRevision(id, signal?)` | 在不加载事件日志的情况下读取一个 id 当前的来源限定修订值。它使用与 `openStored` 相同的修订值表示；id 不存在时返回 `undefined`。 |
 | `appendBatch(meta, events, isMaterialized)` | 持久追加连续批次；尚未实体化时以原子方式延迟实体化。 |
 | `commitRepair(meta, tornMarker, closers)` | 使崩溃修复持久：截断撕裂尾部（当且仅当 `tornMarker !== undefined`；标记可为 falsy，例如 seq/offset `0`），并追加 `closers`。不要求原子性。由 load（截断 + closer）和活动会话接管（仅截断）使用。 |
-| `replaceStored(expectedRevision, meta, events)` | 用完整的当前格式 header 与事件流原子替换一个精确 revision。Revision 与存储身份检查和提交位于同一个 backend 排他区或事务内；不匹配时以 `SessionPersistenceRevisionConflictError` 拒绝。 |
+| `replaceStored(expectedRevision, meta, events)` | 用完整的当前格式 header 与事件流原子替换一个精确 revision。Revision 与存储身份检查发生在提交边界——JSONL 在原子替换前立即检查，SQLite 在替换事务内检查；该检查不提供跨进程写者排他。不匹配时以 `SessionPersistenceRevisionConflictError` 拒绝。 |
 | `list(signal?)` | 列出全部已存储元数据，并遵循可选的取消信号。 |
 | `close?()` | 可选生命周期拆卸（例如关闭 db 句柄），在 dispose drain 后等待其完成。 |

@@ -110,8 +110,9 @@ export interface PersistenceBackend<TornMarker = unknown> {

  /**
   * Atomically replace one exact stored revision with a complete current log.
-   * The backend checks revision and storage identity in the same exclusive file
-   * operation or database transaction that commits the replacement.
+   * The backend checks revision and storage identity at the commit boundary:
+   * immediately before the atomic rename on JSONL, inside the replacing
+   * transaction on SQLite. The check adds no cross-process writer exclusion.
   * @param expectedRevision - exact source revision decoded by the caller.
   * @param meta - complete current-format header.
   * @param events - complete current-format event stream.