fix(runtime): preserve transactions on idempotency conflicts

src/runtime/MonoSystem.ts

@@ -1414,6 +1414,30 @@ RETURNING "lastValue" AS value`,
         this.requireTransaction('dispatchIdempotency.claim')
         const p = this.getPlaceholder()
         const createdAt = Date.now()
+
+        // A unique violation aborts a PostgreSQL transaction. Use a conflict-safe
+        // insert on dialects that support it so the existing row can be read and
+        // translated to the idempotency contract without leaving the transaction
+        // unusable. The dialect is resolved through getSchemaDialect, like every
+        // other dialect decision in this class, so a driver that declares no
+        // schemaDialect is treated as postgres. MySQL has no dispatch transactions
+        // and keeps the unique-error path below.
+        const dialectName = getSchemaDialect(this.db).name
+        const supportsConflictSafeInsert = dialectName === 'postgres' || dialectName === 'sqlite'
+        if (supportsConflictSafeInsert) {
+            const inserted = await this.db.query<{ namespace: string }>(
+                `INSERT INTO "_DispatchIdempotency_" ("namespace", "idempotencyKey", "state", "data", "context", "createdAt")
+                 VALUES (${p()}, ${p()}, 'in_flight', NULL, NULL, ${p()})
+                 ON CONFLICT ("namespace", "idempotencyKey") DO NOTHING
+                 RETURNING "namespace"`,
+                [namespace, idempotencyKey, createdAt],
+                'claim dispatch idempotency row',
+            )
+            if (inserted.length > 0) return
+            // The row already exists; the helper always throws the typed error.
+            return this.rejectDispatchIdempotencyClaim(namespace, idempotencyKey)
+        }
+
         try {
             // Use update (not insert): driver insert always appends RETURNING "_rowId",
             // and this internal table has no _rowId column.
@@ -1427,27 +1451,32 @@ RETURNING "lastValue" AS value`,
         } catch (error) {
             if (normalizeDatabaseError(error, this.db).isUniqueViolation) {
                 // Concurrent claim lost the race — re-load under the same transaction and map.
-                const existing = await this.loadDispatchIdempotencyRow(namespace, idempotencyKey)
-                if (existing?.state === 'succeeded') {
-                    // Caller should have loaded first; treat as conflict rather than replay here.
-                    throw new IdempotencyError({
-                        code: 'IDEMPOTENCY_CONFLICT',
-                        namespace,
-                        idempotencyKey,
-                        causedBy: error instanceof Error ? error : undefined,
-                    })
-                }
-                throw new IdempotencyError({
-                    code: 'IDEMPOTENCY_IN_FLIGHT',
+                await this.rejectDispatchIdempotencyClaim(
                     namespace,
                     idempotencyKey,
-                    causedBy: error instanceof Error ? error : undefined,
-                })
+                    error instanceof Error ? error : undefined,
+                )
             }
             throw error
         }
     }
 
+    private async rejectDispatchIdempotencyClaim(
+        namespace: string,
+        idempotencyKey: string,
+        causedBy?: Error,
+    ): Promise<never> {
+        const existing = await this.loadDispatchIdempotencyRow(namespace, idempotencyKey)
+        throw new IdempotencyError({
+            // Caller should have loaded first; succeeded here means a concurrent
+            // dispatch completed between that read and its claim.
+            code: existing?.state === 'succeeded' ? 'IDEMPOTENCY_CONFLICT' : 'IDEMPOTENCY_IN_FLIGHT',
+            namespace,
+            idempotencyKey,
+            causedBy,
+        })
+    }
+
     private async finishDispatchIdempotencyRow(
         namespace: string,
         idempotencyKey: string,

tests/runtime/WritingComputationTests.md

@@ -320,6 +320,7 @@ test('should handle negative values correctly', async () => {
 | 隐式量化算子的否定语义（r35b 引入） | x:n 路径原子隐含存在量化（∃ 相关行满足谓词），**NOT 与量化子不对易**：¬(∃ 满足) ≠ ∃ 不满足。exist 原子的编译必须保证否定按根记录量化——父路径含 x:n 段时整条路径折叠进 EXISTS（`MatchExp.existAtomCorrelation` 收敛判定：'root' = 反向路径关联根 + 外层零 JOIN；'parent' = 全 x:1 / '&' / 对称段的既有编译），否则外层扇出行逐行求值 NOT EXISTS，「有一个不含匹配终端的中间行」就放行本应排除的根。**EXISTS 关联引用的解析作用域是独立缺陷格**：关联原子必须预解析绑定直接外层查询的别名（`isResolvedFieldReference`，含前缀链）——按 contextRootEntity（最外层根）路径解析在嵌套 EXIST 下绑错表（r25#7 真相）。矩阵：{正向, NOT} × 中段基数 {全 x:1, 1:n, n:n, 双 x:n} × 终段基数 {x:1, x:n} × **中段声明形态 {普通关系, filtered relation}** × {find, update 选择, delete 选择} × 双驱动（`existCorrelationScope.spec.ts`）。**登记边界**：对称段中间路径维持 'parent' 编译（NOT 语义仍按扇出行）；**中段 filtered relation 同样维持 'parent'**（rebased link 谓词 AND 在外层、与路径原子共享 JOIN 别名才有「同一条边」语义——root 折叠会把量化域扩大到全部 base 边，r35 收口自查的邻格探针当场抓获该回归，`hasRebasedPathPredicate` 内部标记收口）；x:n **值原子**（`groups.name = 'G1'`）的 NOT 仍是 SQL 三值 per-row 语义（无组根既不在正集也不在补集）——量化语义未显式声明的面，改动是契约决策不是修复 | r35b（NOT × 多段 exist 误报族 + 嵌套 exist 空集族——存活自 exist 引入；`existJoinPruning.spec.ts` 只铺了正向；filtered 中段 × exist 在既有全量测试中零覆盖） |
 | 运行时派生定义 × 登记面 × 持有 Controller 图（controller-retention 任务引入） | 框架在 setup / migration / storage 编译期合成的派生定义（非用户声明）必须走**唯一非注册构造路径**（`Entity/Relation/Property.derive`，汇合点），不得进入 `X.instances` 用户声明注册表——注册表无移除路径，任何运行时 `X.create()` 合成点都是永久滞留格。取值 = 派生 Klass {Entity / Relation / Property} × 合成路径 {bound-state 属性注入（install 轨 / **migration 轨（`setup(false)` / `createMigrationBaseline` 各自再走一批）**）/ merged 编译（`__type` 判别列 / transformed item / 虚拟 base entity / **虚拟 base relation（仅 filtered-input-into-merged 触发——filtered-over-merged 不触发）** / rebase filtered input）} × 观察面 {注册表长度（经 `KlassByName` 快照）/ `stringifyAllInstances` 纯净性 / `WeakRef` 可回收性 / **闭包捕获物（bound-state `defaultValue` 不得捕获 `RecordBoundState`——它持有 controller；只捕获纯值）**}。**保留链有两族**：Property 直接链与 merged 派生 Entity/Relation 经 clone 图的间接链——只断言一族对另一族失明。**观察时点是判别维度**：断言必须在 setup 后、destroy 前捕获证据（destroy 后观察放过「setup 期登记 + destroy 期反注册」的截断式伪修复）。allowlist spec（`runtimeCreateSiteAllowlist.spec.ts`）把「运行时合成一律走 derive」机器化为源码结构谓词，新增运行时 `X.create()` 站点必须显式决定用户声明 vs 派生定义 | controller-retention（问题陈述只覆盖 Property 直接链；Entity/Relation 间接链由因果实验「只截断 Property 后三 WeakRef 仍存活」揭示——每轮 `Entity+3 / Relation+3+ / Property+9` 登记、Controller/System/Database 永久存活、PGLite 每轮 RSS +≈215 MB） |
 | 迁移期状态图变更 × 作用域 × bound state 所有权 × rebuild 标志（migration-state-machine-state-remap 引入） | StateMachine 的持久化 `currentState` 是**受状态图约束的持久化事实**，不是「有默认值、可整体重置」的附属数据。状态图变更类型 {初始态改名（`stateSignature` 变、双真计划）/ **非初始态改名（`stateSignature` 不变——签名只看默认值，`rebuildState=false`，只看标志会漏）** / 删除节点（removed 无 added 配对）/ 仅新增节点（不需要映射）} × 作用域 {property 级 / global 级} × bound state 所有权 {输出路径不拥有（StateMachine 的 `currentState`、Custom `createState` 的应用语义状态——走映射/重置）/ **输出路径拥有（Transform 的 `sourceRecordId`/`transformIndex`、聚合值与逐项贡献、RealTime 时间戳——`ownsStateOnRebuild` 声明，重置即破坏：键坍缩→唯一索引崩溃或重复行）**} × rebuild 标志组合 {`rebuildState && rebuildOutput` / `rebuildState && !rebuildOutput`（state-only 项在 `if (!item.rebuildOutput) continue` 处跳出——扫描/映射必须挂在 continue 之前）/ `!rebuildState && rebuildOutput`（映射施加以「批准决策存在」为触发面，不能只看标志）}。**验收断言必须双面**：迁移后执行依赖该状态的 Interaction（`error` 为 undefined 且**输出值与 bound state 列都变化**）——只断言迁移计划/handler 返回值放过「迁移成功但下次交互静默 skip」的坏形态；**只改输出值不改状态列的错误实现必须被 currentState 列断言区分**。映射决策组合面（同 dataContext 的 computation 决策）{changed / state-only / **unchanged / unrebuildable（不播种计划→映射被静默丢弃，validateApprovedDiff 必须拒绝）**}。升级窗口（旧 manifest 无 `stateGraph` 字段→「不比较」降级）是独立格：映射要求不产生，合法性不变量（宽口径：rebuild 项触发即扫单列）是唯一防线。样板：`migrationStateMachineStateGraph.spec.ts`（31 用例）+ `postgresqlMigration.spec.ts` 的 `"PostgreSQL state graph mapping migration"` | migration-state-machine-state-remap 机制 A/B/C/D/G（迁移后下一次交互静默无效——`TransitionFinder.findNextState` 对未知名返回 null → `ComputationResult.skip()`，error undefined；PR #51 的两行修复只覆盖机制 A 且放大机制 D 暴露面） |
+| 首写幂等键竞争 × 事务终结 × 隔离级别 × 方言解析 | 两个独立 Controller / PostgreSQL 连接从「同一 `(namespace, key)` 均不存在」开始竞争：首个 dispatch 成功提交时，结果应为一个 `applied`、一个 `replayed`、一次 Interaction resolve 和一条事件；首个 dispatch 在 resolve 失败并回滚时，等待者应插入幂等行并 `applied`，回滚事务不得留下事件或幂等事实。对已存在的 `succeeded` / `in_flight` 行重复调用 claim 时分别映射为 `IDEMPOTENCY_CONFLICT` / `IDEMPOTENCY_IN_FLIGHT`，并在同一事务中继续读取该行。SQLite 与 PGLite 覆盖无竞争的两种现存状态；真实 PostgreSQL 双连接覆盖提交与回滚交错。单 Controller 单连接不作为并发格（驱动将事务串行化）。**隔离级别轴**：READ COMMITTED 是缺陷格（败者的 INSERT 被胜者的未提交行阻塞，胜者提交后被唤醒；旧写法得到唯一冲突并使 PostgreSQL 事务中止）；SERIALIZABLE 是**行为不变的邻格**——败者先经 `load` 读取该键，胜者的 `finish`（带 WHERE 的 UPDATE）也读取该键，PostgreSQL 的可串行化冲突检测在 INSERT 与 `ON CONFLICT DO NOTHING` 两种写法下都报 `40001`，`runWithTransactionRetry` 在新事务中重试后读到已提交结果；该格作为不得回归的固定格，**不能**证明修复本身（旧实现同样通过，只有 READ COMMITTED 格在旧实现上失败）。**方言解析轴**：冲突安全路径由 `getSchemaDialect(db).name`（postgres / sqlite）选择，与 MonoSystem 其余方言判定一致；未声明 `schemaDialect` 的驱动按框架默认视为 postgres，用 `schemaDialect` 为 undefined 的 PGLite 覆盖（裸读 `db.schemaDialect?.name` 的实现在该格失败）。MySQL 无 dispatch 事务，保留唯一冲突路径，不是可达格。业务事务（`runInBusinessTransaction`）内的同 key 竞争不在本格覆盖范围内。 | 同 key 的 PostgreSQL 并发首次 claim：旧实现捕获唯一冲突（SQLSTATE `23505`）后在已中止的事务内继续查询，返回 SQLSTATE `25P02`（`current transaction is aborted`），而不是文档约定的 `replayed` / `IDEMPOTENCY_CONFLICT`；`postgresqlConcurrency.spec.ts` + `dispatchIdempotency.spec.ts` |
 
 **正交轴说明（r19 复盘引入）**：数据形态轴与机制轴都长在"响应式数据流"上（mutation → 事件 → 计算）。
 底层逻辑原语（`BoolExp.evaluate`、match 求值、算术求值）是数据流的**上游依赖**，不产生 mutation、不进事件流、

tests/runtime/dispatchIdempotency.spec.ts

@@ -31,7 +31,7 @@ import {
   Transfer,
   clearAllInstances,
 } from 'interaqt'
-import { PGLiteDB } from '@drivers'
+import { PGLiteDB, SQLiteDB } from '@drivers'
 
 async function tableExists(system: MonoSystem, tableName: string): Promise<boolean> {
   const rows = await (system.storage as any).db.query(
@@ -338,6 +338,73 @@ describe('dispatch idempotency — standalone Interaction', () => {
 
     await system.destroy()
   })
+
+  test('a conflicting claim preserves the transaction and maps the stored state on PGLite, SQLite, and a driver without a declared dialect', async () => {
+    // A driver that declares no schemaDialect is resolved as postgres by getSchemaDialect
+    // (the dialect every other MonoSystem decision uses), so it must take the same
+    // conflict-safe claim path and keep its transaction usable.
+    const createPGLiteWithoutDialect = () => {
+      const database = new PGLiteDB()
+      ;(database as { schemaDialect?: unknown }).schemaDialect = undefined
+      return database
+    }
+    for (const [driver, createDatabase] of [
+      ['PGLite', () => new PGLiteDB()],
+      ['SQLite', () => new SQLiteDB(':memory:')],
+      ['PGLiteWithoutDialect', createPGLiteWithoutDialect],
+    ] as const) {
+      clearAllInstances()
+      const source = Interaction.create({
+        name: `IdemClaimConflict${driver}`,
+        action: Action.create({ name: `idemClaimConflict${driver}` }),
+        idempotency: {
+          key: (args: any) => args.context?.requestId || null,
+        },
+      })
+      const system = new MonoSystem(createDatabase())
+      system.conceptClass = KlassByName
+      const controller = new Controller({ system, eventSources: [source] })
+      await controller.setup(true)
+
+      const applied = await controller.dispatch(source, {
+        user: user(),
+        context: { requestId: 'succeeded' },
+      })
+      expect(applied.error).toBeUndefined()
+      expect(applied.outcome).toBe('applied')
+
+      await system.storage.runInTransaction({ name: `${driver}-seed-inflight` }, async () => {
+        await system.storage.dispatchIdempotency.claim(source.name, 'in-flight')
+      })
+
+      const expectConflictWithReadableRow = async (
+        key: string,
+        expectedCode: 'IDEMPOTENCY_CONFLICT' | 'IDEMPOTENCY_IN_FLIGHT',
+        expectedState: 'succeeded' | 'in_flight',
+      ) => {
+        let conflict: unknown
+        let loadedState: string | undefined
+        await system.storage.runInTransaction({ name: `${driver}-conflicting-claim-${key}` }, async () => {
+          try {
+            await system.storage.dispatchIdempotency.claim(source.name, key)
+          } catch (error) {
+            conflict = error
+          }
+          const stored = await system.storage.dispatchIdempotency.load(source.name, key)
+          loadedState = stored?.state
+        })
+        expect(conflict).toBeInstanceOf(IdempotencyError)
+        expect((conflict as IdempotencyError).code).toBe(expectedCode)
+        expect(loadedState).toBe(expectedState)
+      }
+
+      // The claim conflict is intentionally caught inside its owner transaction:
+      // the subsequent read distinguishes a typed conflict from an aborted SQL transaction.
+      await expectConflictWithReadableRow('in-flight', 'IDEMPOTENCY_IN_FLIGHT', 'in_flight')
+      await expectConflictWithReadableRow('succeeded', 'IDEMPOTENCY_CONFLICT', 'succeeded')
+      await system.destroy()
+    }
+  })
 })
 
 describe('dispatch idempotency — real Activity ★1 / ★4 / I7', () => {