From 42b4a71f7b4ed80ebe7bfbd7a9658e38165cd2c4 Mon Sep 17 00:00:00 2001 From: sskyy Date: Sat, 3 Oct 2026 07:52:57 +0800 Subject: [PATCH 1/2] fix(runtime): preserve transaction on idempotency conflict --- src/runtime/MonoSystem.ts | 53 ++++++--- tests/runtime/WritingComputationTests.md | 1 + tests/runtime/dispatchIdempotency.spec.ts | 60 +++++++++- tests/runtime/postgresqlConcurrency.spec.ts | 115 +++++++++++++++++++- 4 files changed, 213 insertions(+), 16 deletions(-) diff --git a/src/runtime/MonoSystem.ts b/src/runtime/MonoSystem.ts index b3b96ecf..3c57bb13 100644 --- a/src/runtime/MonoSystem.ts +++ b/src/runtime/MonoSystem.ts @@ -1414,6 +1414,26 @@ 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. MySQL and custom dialects retain the unique-error path below. + const supportsConflictSafeInsert = this.db.schemaDialect?.name === 'postgres' + || this.db.schemaDialect?.name === '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 + await 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 +1447,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 { + 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, diff --git a/tests/runtime/WritingComputationTests.md b/tests/runtime/WritingComputationTests.md index 6bab516d..9c21b2e8 100644 --- a/tests/runtime/WritingComputationTests.md +++ b/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 单连接不作为并发格（驱动将事务串行化）。 | C2：同 key 的 PostgreSQL 并发首次 claim 将唯一冲突捕获后在 aborted transaction 内查询，产生 SQLSTATE `25P02`；`postgresqlConcurrency.spec.ts` + `dispatchIdempotency.spec.ts` | **正交轴说明（r19 复盘引入）**：数据形态轴与机制轴都长在"响应式数据流"上（mutation → 事件 → 计算）。 底层逻辑原语（`BoolExp.evaluate`、match 求值、算术求值）是数据流的**上游依赖**，不产生 mutation、不进事件流、 diff --git a/tests/runtime/dispatchIdempotency.spec.ts b/tests/runtime/dispatchIdempotency.spec.ts index e8b445bc..34a5dadd 100644 --- a/tests/runtime/dispatchIdempotency.spec.ts +++ b/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 { const rows = await (system.storage as any).db.query( @@ -338,6 +338,64 @@ describe('dispatch idempotency — standalone Interaction', () => { await system.destroy() }) + + test('a conflicting claim preserves the transaction and maps the stored state on PGLite and SQLite', async () => { + for (const [driver, createDatabase] of [ + ['PGLite', () => new PGLiteDB()], + ['SQLite', () => new SQLiteDB(':memory:')], + ] 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', () => { diff --git a/tests/runtime/postgresqlConcurrency.spec.ts b/tests/runtime/postgresqlConcurrency.spec.ts index 401c2db2..fdc39fbf 100644 --- a/tests/runtime/postgresqlConcurrency.spec.ts +++ b/tests/runtime/postgresqlConcurrency.spec.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from 'vitest'; -import { Entity, Property, Dictionary, Controller, MonoSystem, MatchExp, KlassByName, Relation, Custom, Interaction, Action, Payload, PayloadItem, ComputationResult, runWithTransactionRetry } from 'interaqt'; +import { Entity, Property, Dictionary, Controller, MonoSystem, MatchExp, KlassByName, Relation, Custom, Interaction, InteractionEventEntity, Action, Payload, PayloadItem, ComputationResult, runWithTransactionRetry } from 'interaqt'; import { PostgreSQLDB } from '@drivers'; import { execFile } from 'node:child_process'; import { createRequire } from 'node:module'; @@ -65,6 +65,14 @@ function createBarrier(count: number) { }; } +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise(done => { + resolve = done; + }); + return { promise, resolve }; +} + describeIfPostgres('PostgreSQL computation concurrency', () => { test('allocates unique ids through sequences during concurrent creates', async () => { const sequenceEntity = Entity.create({ @@ -266,6 +274,111 @@ describeIfPostgres('PostgreSQL computation concurrency', () => { } }, 120000); + test('resolves same-key first-claim races across PostgreSQL connections without aborting the loser transaction', async () => { + const source = Interaction.create({ + name: 'pgDispatchIdempotencyRace', + action: Action.create({ name: 'pgDispatchIdempotencyRace' }), + payload: Payload.create({ + items: [PayloadItem.create({ name: 'requestId', type: 'string' })], + }), + idempotency: { + key: (args: any) => args.payload?.requestId || null, + }, + }); + type RaceControl = { + firstResolveEntered: ReturnType>; + releaseFirstResolve: ReturnType>; + failFirstResolve: boolean; + }; + const controls = new Map(); + const resolveCalls = new Map(); + source.resolve = async function(this: Controller, event: any) { + const requestId = String(event.payload.requestId); + const callCount = (resolveCalls.get(requestId) ?? 0) + 1; + resolveCalls.set(requestId, callCount); + const control = controls.get(requestId); + if (callCount === 1 && control) { + control.firstResolveEntered.resolve(); + await control.releaseFirstResolve.promise; + if (control.failFirstResolve) throw new Error(`rollback ${requestId}`); + } + return { requestId, callCount }; + }; + + const system1 = new MonoSystem(new PostgreSQLDB(database, dbOptions)); + const system2 = new MonoSystem(new PostgreSQLDB(database, dbOptions)); + system1.conceptClass = KlassByName; + system2.conceptClass = KlassByName; + const controller1 = new Controller({ system: system1, eventSources: [source] }); + const controller2 = new Controller({ system: system2, eventSources: [source] }); + const claimSignals = new Map>>(); + const idempotency = system2.storage.dispatchIdempotency as any; + const originalClaim = idempotency.claim.bind(idempotency); + idempotency.claim = (namespace: string, key: string) => { + const pending = originalClaim(namespace, key); + if (namespace === source.name) claimSignals.get(key)?.resolve(); + return pending; + }; + + const dispatchRace = async (requestId: string, failFirstResolve: boolean) => { + const control: RaceControl = { + firstResolveEntered: deferred(), + releaseFirstResolve: deferred(), + failFirstResolve, + }; + controls.set(requestId, control); + const loserClaimEntered = deferred(); + claimSignals.set(requestId, loserClaimEntered); + + const firstPromise = controller1.dispatch(source, { + user: { id: 'pg-idempotency-user' }, + payload: { requestId }, + }); + await control.firstResolveEntered.promise; + const secondPromise = controller2.dispatch(source, { + user: { id: 'pg-idempotency-user' }, + payload: { requestId }, + }); + await loserClaimEntered.promise; + control.releaseFirstResolve.resolve(); + const [first, second] = await Promise.all([firstPromise, secondPromise]); + return { first, second }; + }; + + try { + await controller1.setup(true); + await controller2.setup(false); + + const committed = await dispatchRace('winner-commits', false); + expect(committed.first.error).toBeUndefined(); + expect(committed.first.outcome).toBe('applied'); + expect(committed.second.error).toBeUndefined(); + expect(committed.second.outcome).toBe('replayed'); + expect(committed.second.data).toEqual(committed.first.data); + expect(resolveCalls.get('winner-commits')).toBe(1); + + const rolledBack = await dispatchRace('winner-rolls-back', true); + expect(rolledBack.first.error).toBeInstanceOf(Error); + expect(rolledBack.first.outcome).toBeUndefined(); + expect(rolledBack.second.error).toBeUndefined(); + expect(rolledBack.second.outcome).toBe('applied'); + expect(rolledBack.second.data).toEqual({ requestId: 'winner-rolls-back', callCount: 2 }); + expect(resolveCalls.get('winner-rolls-back')).toBe(2); + + const events = await system1.storage.find( + InteractionEventEntity.name!, + MatchExp.atom({ key: 'interactionName', value: ['=', source.name] }), + undefined, + ['*'], + ); + expect(events.filter((event: any) => event.payload?.requestId === 'winner-commits')).toHaveLength(1); + expect(events.filter((event: any) => event.payload?.requestId === 'winner-rolls-back')).toHaveLength(1); + } finally { + for (const control of controls.values()) control.releaseFirstResolve.resolve(); + await Promise.all([system1.destroy(), system2.destroy()]); + } + }, 120000); + test('runs concurrent serializable custom dispatches through retry', async () => { const Counter = Entity.create({ name: 'PgSerializableDispatchCounter', From 58a610ae1878637d9a136b4b22e9b5679ebec795 Mon Sep 17 00:00:00 2001 From: sskyy Date: Sat, 3 Oct 2026 22:38:23 +0800 Subject: [PATCH 2/2] fix(runtime): resolve the claim dialect through getSchemaDialect and pin neighbouring cells The conflict-safe idempotency claim chose its path from the raw `db.schemaDialect?.name`, while every other dialect decision in MonoSystem uses `getSchemaDialect(db).name`, which treats a driver without a declared dialect as postgres. A PostgreSQL-compatible driver that declares no dialect therefore kept the aborted-transaction defect. The claim now resolves the dialect the same way, and returns the typed rejection explicitly instead of relying on fall-through. Tests: add the no-declared-dialect driver cell (fails with the raw read), and a real-PostgreSQL SERIALIZABLE cell. The SERIALIZABLE cell behaves the same before and after the change (the winner's finish() reads the key, so PostgreSQL reports 40001 for both INSERT forms); it is registered as a pinned neighbouring cell, not as a reproduction of the defect. Registry row now names the isolation and dialect-resolution axes and drops the undefined label. Co-Authored-By: Claude Sonnet 5.5 --- src/runtime/MonoSystem.ts | 12 ++- tests/runtime/WritingComputationTests.md | 2 +- tests/runtime/dispatchIdempotency.spec.ts | 11 ++- tests/runtime/postgresqlConcurrency.spec.ts | 82 ++++++++++++++++++++- 4 files changed, 100 insertions(+), 7 deletions(-) diff --git a/src/runtime/MonoSystem.ts b/src/runtime/MonoSystem.ts index 3c57bb13..42b55fda 100644 --- a/src/runtime/MonoSystem.ts +++ b/src/runtime/MonoSystem.ts @@ -1418,9 +1418,12 @@ RETURNING "lastValue" AS value`, // 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. MySQL and custom dialects retain the unique-error path below. - const supportsConflictSafeInsert = this.db.schemaDialect?.name === 'postgres' - || this.db.schemaDialect?.name === 'sqlite' + // 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") @@ -1431,7 +1434,8 @@ RETURNING "lastValue" AS value`, 'claim dispatch idempotency row', ) if (inserted.length > 0) return - await this.rejectDispatchIdempotencyClaim(namespace, idempotencyKey) + // The row already exists; the helper always throws the typed error. + return this.rejectDispatchIdempotencyClaim(namespace, idempotencyKey) } try { diff --git a/tests/runtime/WritingComputationTests.md b/tests/runtime/WritingComputationTests.md index 9c21b2e8..f3c1be4c 100644 --- a/tests/runtime/WritingComputationTests.md +++ b/tests/runtime/WritingComputationTests.md @@ -320,7 +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 单连接不作为并发格（驱动将事务串行化）。 | C2：同 key 的 PostgreSQL 并发首次 claim 将唯一冲突捕获后在 aborted transaction 内查询，产生 SQLSTATE `25P02`；`postgresqlConcurrency.spec.ts` + `dispatchIdempotency.spec.ts` | +| 首写幂等键竞争 × 事务终结 × 隔离级别 × 方言解析 | 两个独立 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、不进事件流、 diff --git a/tests/runtime/dispatchIdempotency.spec.ts b/tests/runtime/dispatchIdempotency.spec.ts index 34a5dadd..03e8bd34 100644 --- a/tests/runtime/dispatchIdempotency.spec.ts +++ b/tests/runtime/dispatchIdempotency.spec.ts @@ -339,10 +339,19 @@ describe('dispatch idempotency — standalone Interaction', () => { await system.destroy() }) - test('a conflicting claim preserves the transaction and maps the stored state on PGLite and SQLite', async () => { + 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({ diff --git a/tests/runtime/postgresqlConcurrency.spec.ts b/tests/runtime/postgresqlConcurrency.spec.ts index fdc39fbf..8734938e 100644 --- a/tests/runtime/postgresqlConcurrency.spec.ts +++ b/tests/runtime/postgresqlConcurrency.spec.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from 'vitest'; -import { Entity, Property, Dictionary, Controller, MonoSystem, MatchExp, KlassByName, Relation, Custom, Interaction, InteractionEventEntity, Action, Payload, PayloadItem, ComputationResult, runWithTransactionRetry } from 'interaqt'; +import { Entity, Property, Dictionary, Controller, MonoSystem, MatchExp, KlassByName, Relation, Custom, Interaction, InteractionEventEntity, Action, Payload, PayloadItem, ComputationResult, runWithTransactionRetry, isRetryableTransactionError } from 'interaqt'; import { PostgreSQLDB } from '@drivers'; import { execFile } from 'node:child_process'; import { createRequire } from 'node:module'; @@ -379,6 +379,86 @@ describeIfPostgres('PostgreSQL computation concurrency', () => { } }, 120000); + test('keeps the retryable serialization failure of a SERIALIZABLE claim that loses to a first claim', async () => { + const source = Interaction.create({ + name: 'pgDispatchIdempotencySerializableClaim', + action: Action.create({ name: 'pgDispatchIdempotencySerializableClaim' }), + idempotency: { + key: (args: any) => args.context?.requestId || null, + }, + }); + const key = 'serializable-claim'; + const system1 = new MonoSystem(new PostgreSQLDB(database, dbOptions)); + const system2 = new MonoSystem(new PostgreSQLDB(database, dbOptions)); + system1.conceptClass = KlassByName; + system2.conceptClass = KlassByName; + const controller1 = new Controller({ system: system1, eventSources: [source] }); + const controller2 = new Controller({ system: system2, eventSources: [source] }); + const winnerFinished = deferred(); + const releaseWinner = deferred(); + + try { + await controller1.setup(true); + await controller2.setup(false); + + // The winner claims and finishes the key but has not committed yet. + const winner = system1.storage.runInTransaction({ name: 'pg-serializable-claim-winner', isolation: 'SERIALIZABLE' }, async () => { + await system1.storage.dispatchIdempotency.claim(source.name, key); + await system1.storage.dispatchIdempotency.finish(source.name, key, { data: { winner: true }, createdAt: Date.now() }); + winnerFinished.resolve(); + await releaseWinner.promise; + }); + await winnerFinished.promise; + + // Neighbouring cell of the READ COMMITTED race above: this cell behaves the same before + // and after the conflict-safe insert, and is pinned so it cannot regress. The winner's + // finish() reads the key after inserting it, and the loser read the key through load(), + // so PostgreSQL's serializable conflict detection sees a read/write cycle and reports + // 40001 for a plain INSERT and for ON CONFLICT DO NOTHING alike. The framework retries + // 40001 in a new transaction whose snapshot contains the committed row, so the loser + // ends up reading the winner's result. The loser claims while the winner is still + // uncommitted, so its INSERT is blocked on the winner when the winner commits. + const waitUntilClaimIsBlocked = async () => { + const observer = (system1.storage as any).db; + for (let poll = 0; poll < 100; poll++) { + const rows = await observer.query( + `SELECT count(*)::int AS "blocked" FROM pg_stat_activity + WHERE datname = current_database() AND wait_event_type = 'Lock' + AND query LIKE '%_DispatchIdempotency_%'`, + [], + 'observe blocked idempotency claim', + ); + if (Number(rows[0].blocked) > 0) return; + await new Promise(resolve => setTimeout(resolve, 50)); + } + throw new Error('the loser claim never blocked on the winner'); + }; + let firstAttemptError: unknown; + const outcome = await runWithTransactionRetry('pg-serializable-claim-loser', async (_isolation, attempt) => + system2.storage.runInTransaction({ name: 'pg-serializable-claim-loser', isolation: 'SERIALIZABLE' }, async () => { + const stored = await system2.storage.dispatchIdempotency.load(source.name, key); + if (stored) return { attempt, state: stored.state, data: stored.data }; + expect(attempt).toBe(1); + const claimResult = system2.storage.dispatchIdempotency.claim(source.name, key).then(() => undefined, (error: unknown) => error); + await waitUntilClaimIsBlocked(); + releaseWinner.resolve(); + await winner; + firstAttemptError = await claimResult; + if (!firstAttemptError) throw new Error('claim unexpectedly succeeded'); + throw firstAttemptError; + }) + ); + + expect(firstAttemptError).toBeDefined(); + expect((firstAttemptError as { code?: string }).code).toBe('40001'); + expect(isRetryableTransactionError(firstAttemptError)).toBe(true); + expect(outcome).toEqual({ attempt: 2, state: 'succeeded', data: { winner: true } }); + } finally { + releaseWinner.resolve(); + await Promise.all([system1.destroy(), system2.destroy()]); + } + }, 120000); + test('runs concurrent serializable custom dispatches through retry', async () => { const Counter = Entity.create({ name: 'PgSerializableDispatchCounter',