事务、并发与幂等

通过额度预留理解原子更新、事务边界与幂等记录，处理重复请求和并发修改而不依赖前端按钮防抖。

L2 · 能交付约 20 分钟阅读含示例、练习与验收

建议先读：PostgreSQL 数据建模 SQL、索引与查询计划异步、并发与取消

本页内容

本章解决什么问题#

用户点击“解析文档”后网络超时，重新点击又扣了一次额度；两个请求同时读到剩余一百，都认为足够，各自扣八十，最终状态却违反规则。这些问题靠禁用按钮不能解决，因为重试也可能来自代理、任务队列或另一个服务实例。本章目标是用事务、条件更新和幂等记录保护一次业务操作，清楚说明失败时哪些效果已发生。前置是参数化 SQL、连接以及业务不变量。

前端的 optimistic update 是先更新显示再等服务器确认；数据库事务是把若干状态变化组成一个提交单元。两者不能互相代替。界面回滚只是恢复用户看到的状态，后端仍必须保证额度与解析任务不会出现一边成功、一边失败的残缺状态。

三种保证分别解决什么#

原子性意味着一个事务里的数据库修改要么一起提交，要么回滚；隔离性控制同时进行的事务能看到什么；持久性涉及已经确认提交的结果如何保留。这里先把精力放在业务边界：额度预留与幂等结果应属于同一个事务，模型网络调用不应因为“看起来相关”就放进持锁事务中等待几分钟。

PostgreSQL 默认 Read Committed 中，每条语句会获得当时可见的已提交数据视图，同一个事务两次查询可能看到不同结果。先 SELECT balance，再由 JavaScript 计算并 UPDATE，是典型的读改写竞争。对于“余额足够才扣减”，可以直接用 UPDATE ... SET balance = balance - amount WHERE balance >= amount RETURNING balance，让检查与变更发生在同一条受数据库并发控制保护的语句内。事务隔离文档解释了这一层的可见性规则。

幂等则回答重复请求的效果。客户端生成一个业务请求键，服务端在“租户加键”的范围内保存请求摘要和结果；同一个键、相同参数返回原结果，同一个键、不同参数明确冲突。只缓存一个“处理过”标记而不保存结果，会让超时重试无法判断到底扣了多少。键的有效期、作用域和清理策略都属于 API 合同。

核心 SQL 与连接合同#

BEGIN 开始事务，COMMIT 提交，ROLLBACK 撤销尚未提交的修改。使用 pg 的 Pool 时，必须先 pool.connect() 获取 Client，事务中的所有 SQL 都用这个 Client。不能连续调用 pool.query('BEGIN')、pool.query(...)、pool.query('COMMIT') 期待它们落在同一条连接上。node-postgres 事务文档对此有明确要求。

INSERT ... ON CONFLICT DO NOTHING RETURNING ... 可通过唯一约束争夺某个幂等键。首次插入返回一行，重复时返回零行；唯一约束负责处理并发竞争，应用不应先查询不存在再盲目插入。RETURNING 直接返回当前语句修改后的数据，避免为了拿新余额再写一次可能看到别的状态的查询。

错误代码 40001 表示序列化失败，40P01 表示死锁，某些事务可整体重试；不能只重试最后一条 UPDATE。提交时连接中断可能让客户端不知道事务是否提交，因此幂等键还承担恢复查询的作用。ROLLBACK 也可能失败，生产代码应保留原始错误、记录回滚错误并丢弃不可再用的连接。

完整示例：一次扣额，多次读取同一结果#

环境：Node 22.22 或 24、PostgreSQL 16+、pg 8。空练习目录执行 npm.cmd init -y、npm.cmd install pg@8，设置自己的 DATABASE_URL。PowerShell 示例为 $env:DATABASE_URL='postgresql://course:course_password@127.0.0.1:5432/ai_course'。保存 transactions.mjs 后运行 node transactions.mjs。示例在一条连接的临时表上验证重放、冲突和回滚；它没有伪装成已经执行过多连接并发压力测试。

transactions.mjs

// transactions.mjs
import pg from 'pg';
import assert from 'node:assert/strict';

if (!process.env.DATABASE_URL) throw new Error('请设置 DATABASE_URL');
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL, max: 2 });
pool.on('error', (error) => console.error('空闲连接错误：', error.message));
const client = await pool.connect();
let broken = false;

async function reserveCredits(db, { tenantId, key, accountId, amount }) {
  if (!Number.isInteger(amount) || amount < 1 || amount > 1000000) {
    throw new RangeError('amount 不合法');
  }
  if (typeof key !== 'string' || key.length < 8 || key.length > 100) {
    throw new RangeError('幂等键长度不合法');
  }
  await db.query('BEGIN');
  try {
    const inserted = await db.query(`INSERT INTO requests
      (tenant_id, request_key, account_id, amount)
      VALUES ($1, $2, $3, $4)
      ON CONFLICT (tenant_id, request_key) DO NOTHING RETURNING request_key`,
      [tenantId, key, accountId, amount]);
    if (inserted.rowCount === 0) {
      const old = (await db.query(`SELECT account_id, amount, response
        FROM requests WHERE tenant_id = $1 AND request_key = $2`,
        [tenantId, key])).rows[0];
      if (!old || old.account_id !== accountId || old.amount !== amount) {
        throw Object.assign(new Error('相同键对应了不同请求'), { code: 'KEY_CONFLICT' });
      }
      await db.query('COMMIT');
      return { ...old.response, replayed: true };
    }

    const changed = await db.query(`UPDATE accounts
      SET balance = balance - $3
      WHERE tenant_id = $1 AND id = $2 AND balance >= $3
      RETURNING balance`, [tenantId, accountId, amount]);
    if (changed.rowCount !== 1) {
      throw Object.assign(new Error('账户不存在或额度不足'), { code: 'NO_CREDIT' });
    }
    const response = { remaining: changed.rows[0].balance, charged: amount };
    await db.query(`UPDATE requests SET response = $3::jsonb
      WHERE tenant_id = $1 AND request_key = $2`,
      [tenantId, key, JSON.stringify(response)]);
    await db.query('COMMIT');
    return { ...response, replayed: false };
  } catch (error) {
    try { await db.query('ROLLBACK'); }
    catch (rollbackError) {
      broken = true;
      console.error('回滚失败：', rollbackError.message);
    }
    throw error;
  }
}

try {
  await client.query(`
    CREATE TEMP TABLE accounts (
      tenant_id integer NOT NULL, id integer NOT NULL,
      balance integer NOT NULL CHECK (balance >= 0),
      PRIMARY KEY (tenant_id, id)
    );
    CREATE TEMP TABLE requests (
      tenant_id integer NOT NULL, request_key text NOT NULL,
      account_id integer NOT NULL, amount integer NOT NULL CHECK (amount > 0),
      response jsonb,
      PRIMARY KEY (tenant_id, request_key)
    );
    INSERT INTO accounts VALUES (1, 10, 100);
  `);
  const input = { tenantId: 1, key: 'request-0001', accountId: 10, amount: 30 };
  const first = await reserveCredits(client, input);
  const repeated = await reserveCredits(client, input);
  assert.equal(first.remaining, 70);
  assert.equal(repeated.remaining, 70);
  assert.equal(repeated.replayed, true);
  await assert.rejects(() => reserveCredits(client, { ...input, amount: 31 }),
    { code: 'KEY_CONFLICT' });
  await assert.rejects(() => reserveCredits(client,
    { ...input, key: 'request-0002', amount: 90 }), { code: 'NO_CREDIT' });
  const balance = (await client.query('SELECT balance FROM accounts')).rows[0].balance;
  assert.equal(balance, 70);
  console.log('只扣除一次：剩余额度 70；参数冲突与余额不足均已回滚');
} finally {
  client.release(broken);
  await pool.end();
}

预期输出只扣除一次的说明。数据库必须由读者准备，连接失败不能跳过。示例返回的 replayed 表示服务器是否复用了先前结果，不代表业务再次执行。应用可以选择不对外暴露这个调试字段，但需保持返回内容合同一致。

关键代码逐段解释#

幂等记录与扣额位于同一事务：扣额失败会连同首次插入的请求记录一起回滚，不留下永远“处理中”的空壳。重复请求先被唯一约束挡住，再查询已经提交的请求内容；示例使用默认隔离级别，另一事务插入相同键时可能发生等待，不能把这种等待误判为应用死循环。

条件 UPDATE 避免在应用中读出旧余额再覆盖。rowCount 为零表示本次没有扣成功，不能只看 SQL 没有抛异常就返回成功。余额下限的 CHECK 再提供一层防护，但它不能单独告诉产品失败应显示什么，因此业务层仍要翻译错误。

本例在一个 Client 上依次执行事务。真实服务每次操作应独占获取一条连接，不能在同一个 Client 上 Promise.all 启动两个事务，否则语句边界会交错。连接池大小也不是越大越好，所有服务实例的最大连接数总和应与数据库容量匹配。

深入阅读：先写出一次提交必须保护的事实#

在“团队文档与任务助手”中，点击解析通常要产生三种事实：团队额度减少、解析任务出现、这次请求的结果可以再次查询。不要先问应该开几个事务，先把业务句子写完整：“只要页面获得了已接受的任务编号，数据库里就必须存在对应任务；如果创建任务失败，不能减少额度；同一次用户意图被网络重复传输时，最多减少一次额度。”这些句子是事务设计的输入，表名只是实现材料。

三个事实还不够描述全部流程。模型实际执行可能在一分钟以后失败，此时是否返还额度，是另一项业务决策。可以预留后结算，也可以先扣后退款，但都应有对应记录，不能在失败时悄悄把 balance 加回去就认为账目已经解释清楚。用户重新解析同一文档，也可能是新的付费操作，所以不能永远以 document_id 去重。幂等键标识一次请求意图，文档编号标识业务资源，两者不能混用。

前端的 loading 锁只能阻止当前组件重复点击，无法保护另一个浏览器窗口，也无法保护客户端已发出但未收到响应的请求。服务端内存锁同样只能保护一个进程。真正被多个实例共享的余额必须在共享存储上建立竞争规则。把业务不变量放到唯一约束、检查约束和条件更新中，才不会因为新增一个消费者或扩容一个实例就消失。

API 细读：事务命令、连接与返回结果#

pool.connect() 没有业务参数，返回 Promise。它可能复用空闲连接，也可能建立新连接，还可能等待容量；等待多久取决于连接池配置，不能假定马上获得连接。连接失败、认证失败和等待超时都发生在业务 SQL 之前。获得 Client 后必须明确所有权：本次业务操作独占使用，最后释放。client.release() 把健康连接交回池，传入真值可销毁有问题的连接；释放以后继续用它查询，会破坏所有权约定。

client.query('BEGIN') 的默认事务隔离级别由数据库配置决定，普通安装通常是 Read Committed。本章不会把应用配置与数据库默认值混为一谈：需要固定语义时，可以在开始事务时明确声明级别。COMMIT 成功代表数据库已接受提交，客户端在提交等待中掉线则属于结果不确定；不能把捕获到网络错误直接解释成数据库一定回滚。ROLLBACK 只能撤销尚未提交的数据库修改，不能取消已发出的 HTTP 请求。

query(text, values) 的 values 默认为未提供；传入占位符就应按顺序提供对应参数。返回的 rows 是结果行数组，rowCount 是命令实际影响的行数，它们不是同一个概念。没有 RETURNING 的 UPDATE 可能 rowCount 为一但 rows 为空。把 rows.length 等于零理解为更新失败，是把“没有要求返回记录”误读成“没有修改记录”。本章用 rowCount 判断是否真正扣额，用 RETURNING 读取这次扣额的新余额。

ON CONFLICT 需要数据库有与冲突目标相匹配的唯一约束或唯一索引，否则 SQL 本身就无法执行。DO NOTHING 不会覆盖旧请求参数，因此随后必须读旧记录并比较语义字段。请求摘要应从稳定、明确的字段编码生成，不能盲目对任意请求对象 JSON.stringify：字段顺序、无关追踪字段或不同数值表示可能改变摘要。对于本章固定 DTO，直接比较账户、文档和额度最容易解释。

最小实验：亲眼看到读改写丢失#

先运行不需要数据库的实验，确认问题发生在时间交错，而不是某种 ORM 写法。保存为 transaction-race.mjs，Node 22.22 或 24 执行 node transaction-race.mjs，无依赖。这里用共享变量模拟余额，仅说明竞争机制，不代替 PostgreSQL 隔离级别的实测。

transaction-race.mjs

// transaction-race.mjs
import assert from 'node:assert/strict';
import { setTimeout as delay } from 'node:timers/promises';

let balance = 100;
async function brokenDebit(amount) {
  const observed = balance; // 两个调用都能在等待前读到 100。
  if (observed < amount) return false;
  await delay(10);
  balance = observed - amount;
  return true;
}
const broken = await Promise.all([brokenDebit(80), brokenDebit(80)]);
assert.deepEqual(broken, [true, true]);
assert.equal(balance, 20);
console.log('错误实现：两次都报告扣 80，但余额只减少了 80');

balance = 100;
function singleStepDebit(amount) {
  // 没有 await，检查与修改在同一 JavaScript 执行片段内完成。
  if (balance < amount) return false;
  balance -= amount;
  return true;
}
const guarded = [singleStepDebit(80), singleStepDebit(80)];
assert.deepEqual(guarded, [true, false]);
assert.equal(balance, 20);
console.log('单步模型：一次成功，一次余额不足，余额为 20');

预期是两条说明。第一条揭示的不是余额变负，而是记录与实际扣款不一致：两位调用方都以为成功，系统只保存了一次减少。这提醒我们不能只检查最终余额非负，还要核对成功操作数和业务记录。第二种写法只在这个单进程同步模型内不可交错；把它移到两个服务器实例不会共享同一变量。PostgreSQL 条件 UPDATE 把类似的检查与变更交给数据库协调，才覆盖跨实例竞争。

沿着两个请求推演幂等路径#

请求甲先插入键 K，但还没提交；请求乙也插入 K。唯一约束不会让两行都长期存在，乙可能等待甲的结果。如果甲完成扣额并提交，乙发现冲突，读取甲保存的结果并返回。如果甲在创建任务时失败并回滚，甲插入的键也消失，乙就有机会成为真正执行者。应用不能用一个提前提交的“已处理”标记替代这个整体边界，否则甲失败后乙可能永远被拦住。

默认 Read Committed 下，乙的后续 SELECT 是另一条语句，会获得新的可见性视图，因此可以读取已经提交的请求记录。换到更强隔离级别，某些并发路径可能直接报序列化失败；实现就必须准备整体重试。更强隔离不是“把配置改高就不用考虑并发”，它往往把潜在异常转换成显式冲突，让应用有机会重试或告知用户。

锁的顺序也属于设计。一次转账同时更新两份账户时，如果甲先锁 A 后锁 B，乙先锁 B 后锁 A，就可能彼此等待。统一按账户编号排序获取锁可以减少这种死锁，仍应处理数据库报出的死锁错误。需要重试时，从事务起点重新读取数据、重新判断条件、重新执行，而不是在失败事务里只补发最后一条语句。

把原练习落成完整参考实现#

下面实现原练习“扣额与创建解析作业一起成功”。保存为 quota-job.mjs，安装 pg 8 并配置专用练习库的 DATABASE_URL，执行 node quota-job.mjs。代码先故意在扣额后失败，再使用同一请求键重试，最后验证只创建一个任务。所有表是临时表；不需要改动现有业务表。真实接口传入的租户与账户必须来自授权层，而不是直接信任请求体。

完整代码已收录在本章末尾的练习参考答案中；可先阅读说明，再展开复制运行。

调试观察：定位等待、失败和不确定结果#

出现“接口一直转圈”时，先区分获取连接等待、SQL 执行等待和外部网络等待。每个阶段单独记录开始时间、结束时间与 request_id，日志中保存请求键的安全标识和业务编号，不记录密码或完整文档。只有一个总耗时，无法判断到底应调整池容量、索引还是供应商超时。数据库侧可以观察活动会话与锁等待，但生产环境读取监控视图也应使用合适权限，避免直接取消不了解归属的查询。

重现并发问题要设计时间线，而不是靠快速连点。练习库中可以让两个连接先读同一值，再用可控屏障同时继续；断言除了最终余额，还应包括成功请求数、任务数和幂等记录数。临时表只属于建立它的连接，不能拿本章临时表脚本直接用另一个 Client 并发查询；需要先建立独立持久练习表并明确清理范围，这也是本稿没有把顺序自检说成并发验收的原因。

真实项目接入与取舍#

HTTP 层负责解析和验证幂等键、构建可信租户上下文；服务层拥有事务边界；仓储函数接收当前 Client，而不是各自偷偷从全局 pool 获取连接。这样创建任务、扣额和保存响应才能共享提交。可以把它封装成 withTransaction，但封装应保留返回值、原始错误与连接清理语义，不要为了“统一错误”把序列化冲突和业务余额不足都吞成一个布尔值。

团队文档助手在调用模型前，适合先提交额度预留和 pending 作业；工作者成功后结算用量，失败后按产品策略释放预留。外部费用可能与本地估算不同，最好保存预估、实际和调整记录。系统越接近计费，越需要可追踪的流水，而不是只保存一个会变化的余额。业务量小的时候，一份清楚的 PostgreSQL 模型通常比同时引入多个锁服务更容易验证；规模变化后再根据实际竞争与延迟选择更复杂的方案。

更进一步：跨行规则为什么需要另一种思考#

单行条件更新很适合扣减一份余额，但并不是所有规则都能压进一条余额记录。假设团队规定同时最多运行三个高成本解析任务，现在已有两个。请求甲查询运行数为二，请求乙也查询为二，然后各插入一个任务，最终变成四个。每次插入本身都合法，任务表里也没有重复主键；问题出在两个事务共同依赖了一个跨行判断。只给每个任务主键加锁不会自动保护“总数不能超过三”这条规则。

一种设计是把并发配额放到团队资源记录中，用条件更新预留名额，再创建任务；释放名额也必须成为作业完成或补偿的一部分。另一种设计是对同一个团队控制行加锁，在锁内重新统计并决定是否接受；这会把同团队的操作串行化，可能影响吞吐。还可以采用 Serializable 并处理整个事务重试。选择依据是竞争频率、规则范围与可解释性，而不是笼统地认为某个隔离级别“更高级”。

这也说明冗余计数不是天然错误。它把难以锁定的集合规则集中为一个可竞争的资源，但需要修复机制：统计任务表与计数差异，发现异常时记录证据并按照明确流程校正。真实系统同时考虑正常写入路径与异常恢复路径，不能只证明一次成功请求能走通。

为什么提交数据库后立即发队列仍会丢任务#

常见写法是先 COMMIT，再调用队列 SDK。数据库提交成功后进程若在 SDK 调用前退出，任务已经存在但没有消息；反过来先发消息再提交，消费者可能收到一条对应事务最终回滚的消息。这不是交换两行代码就能消除的问题，因为数据库与外部队列有各自的提交边界。

outbox 的做法是在业务事务里同时写一条“待发送事件”。提交后，转发器查询这些事件，向外部队列投递，再记录已投递。转发器也可能在投递成功、记录状态之前崩溃，因此同一事件仍可能重复发送；消费者使用事件编号或业务键去重。它把“可能永远没发出去”转换成“允许重复但可恢复”，并没有凭空创造端到端恰好一次执行。

对于团队文档助手，事件负载通常保存任务编号、租户编号和版本，正文保留在受控存储中。消费者执行前读取当前任务，确认版本与状态，避免队列里塞入很大的私有文档副本。消息保留期限、用户删除请求与审计保留之间的关系也应提前决定，尤其不能让业务删除了文档，队列副本却继续被任意重放。

幂等结果应保存多久#

幂等键不是无限期免费缓存。保存太短，移动网络中的延迟重试可能在记录清理后再次执行；保存太久，则增加存储与隐私管理成本。接口文档应说明有效期，客户端在期限内重试沿用原键，超过期限先查询业务结果，不自动生成新键扣费。任务记录和扣额流水最好具有独立业务唯一标识，让短期幂等缓存清理后仍能追查一次操作。

还要区分“接口参数相同”与“用户意图相同”。用户第一次解析文档版本三，第二次主动要求重新解析版本四，即使标题没有变化，也应是不同业务意图。请求记录应包含文档版本，结果应返回对应任务编号。前端因此可以在重连后恢复正确任务，而不是拿旧摘要覆盖刚刚上传的新版本。

边界、练习与参考解答#

数据库事务不能回滚已经发给模型供应商的请求、邮件或支付操作。可靠方式常是事务里写业务状态和待执行事件，提交后由后台任务处理；外部副作用还要使用自己的幂等键。不要持有余额锁等待几分钟模型输出。对提交结果不确定的请求，客户端应沿用原键查询或重试，不能换新键重新扣额。

练习：在额度预留成功后同时创建解析作业，要求任何一步失败都不扣额。提示：新增 jobs 表，把 INSERT 放在同一个事务里；发送到外部队列不属于本地 SQL 事务。

参考答案（含完整可运行实现）

在条件扣额成功后插入带 tenant_id、document_id、request_key 的 jobs，保存 job_id 到幂等 response，再提交。给 jobs 的业务去重字段建立唯一约束，并让后台工作者读取 committed 的待处理作业。若实际使用外部队列，则加入 outbox 记录，由转发器提交后投递；转发可以重复，消费者必须幂等。通过故意触发 jobs 约束失败，验证余额、请求记录和作业都没有留下部分修改。完整可运行参考代码见后面的“额度与作业一并提交”。

quota-job.mjs

// quota-job.mjs
import pg from 'pg';
import assert from 'node:assert/strict';
import { randomUUID } from 'node:crypto';
if (!process.env.DATABASE_URL) throw new Error('请设置 DATABASE_URL');
const db = new pg.Client({ connectionString: process.env.DATABASE_URL });
await db.connect();

async function enqueue(input, failAfterDebit = false) {
  const { tenantId, accountId, documentId, cost, key } = input;
  if (!Number.isInteger(cost) || cost < 1 || cost > 1000) throw new Error('额度不合法');
  if (typeof key !== 'string' || key.length < 8 || key.length > 100) throw new Error('请求键不合法');
  await db.query('BEGIN');
  try {
    const inserted = await db.query(`INSERT INTO ex_requests
      (tenant_id, key, account_id, document_id, cost)
      VALUES ($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING RETURNING key`,
      [tenantId, key, accountId, documentId, cost]);
    if (!inserted.rowCount) {
      const old = (await db.query(`SELECT account_id,document_id,cost,result
        FROM ex_requests WHERE tenant_id=$1 AND key=$2`, [tenantId,key])).rows[0];
      if (!old || old.account_id !== accountId || old.document_id !== documentId || old.cost !== cost) {
        throw new Error('请求键冲突');
      }
      await db.query('COMMIT');
      return old.result;
    }
    const changed = await db.query(`UPDATE ex_accounts SET balance=balance-$3
      WHERE tenant_id=$1 AND id=$2 AND balance >= $3 RETURNING balance`,
      [tenantId,accountId,cost]);
    if (changed.rowCount !== 1) throw new Error('额度不足');
    if (failAfterDebit) throw new Error('模拟作业创建前故障');
    const jobId = randomUUID();
    await db.query(`INSERT INTO ex_jobs (id,tenant_id,document_id,status)
      VALUES ($1,$2,$3,'pending')`, [jobId,tenantId,documentId]);
    const result = { jobId, remaining: changed.rows[0].balance };
    await db.query(`UPDATE ex_requests SET result=$3::jsonb
      WHERE tenant_id=$1 AND key=$2`, [tenantId,key,JSON.stringify(result)]);
    await db.query('COMMIT');
    return result;
  } catch (error) {
    try { await db.query('ROLLBACK'); }
    catch (rollbackError) { throw new AggregateError([error,rollbackError], '事务与回滚都失败'); }
    throw error;
  }
}
try {
  await db.query(`
    CREATE TEMP TABLE ex_accounts (tenant_id integer,id integer,balance integer CHECK(balance>=0),
      PRIMARY KEY(tenant_id,id));
    CREATE TEMP TABLE ex_documents (tenant_id integer,id integer,PRIMARY KEY(tenant_id,id));
    CREATE TEMP TABLE ex_requests (tenant_id integer,key text,account_id integer,
      document_id integer,cost integer NOT NULL CHECK(cost>0),result jsonb,PRIMARY KEY(tenant_id,key));
    CREATE TEMP TABLE ex_jobs (id uuid PRIMARY KEY,tenant_id integer NOT NULL,
      document_id integer NOT NULL,status text NOT NULL CHECK(status IN ('pending','done')),
      FOREIGN KEY(tenant_id,document_id) REFERENCES ex_documents(tenant_id,id));
    INSERT INTO ex_accounts VALUES(1,10,100);
    INSERT INTO ex_documents VALUES(1,42);
  `);
  const input = {tenantId:1,accountId:10,documentId:42,cost:30,key:'parse-request-0001'};
  await assert.rejects(() => enqueue(input,true), /模拟作业/);
  assert.equal((await db.query('SELECT balance FROM ex_accounts')).rows[0].balance,100);
  assert.equal((await db.query('SELECT count(*)::int AS n FROM ex_requests')).rows[0].n,0);
  const first = await enqueue(input);
  const again = await enqueue(input);
  assert.equal(again.jobId,first.jobId);
  assert.equal(again.remaining,70);
  assert.equal((await db.query('SELECT count(*)::int AS n FROM ex_jobs')).rows[0].n,1);
  console.log('故障未扣额；重试得到同一个任务；最终余额 70、任务数 1');
} finally { await db.end(); }

可验证的验收标准#

相同键与参数多次请求只扣一次；相同键不同参数被拒绝；余额不足后余额和幂等记录一起回滚；事务所有语句使用同一 Client；连接最终归还。扩展到多连接实验时，必须在独立练习表上验证两个竞争请求的总扣额，而不能把本例顺序重放当成并发证据。

三个自测问题与答案#

为什么不能用 pool.query 依次发送事务语句？答案：不同调用可能分配不同连接，无法形成同一事务。
事务失败时能只重试最后一条语句吗？答案：通常不能，事务可能已进入失败状态或读取前提已变化，应按可重试策略整体重做。
前端收到超时后换一个幂等键是否安全？答案：可能重复执行，应保留原键以恢复同一次业务操作的结果。

本章验证记录#

编写时使用 Node 22.22.0 对本章全部 3 个 JavaScript 完整文件执行了语法检查。已在本机实际运行通过：transaction-race.mjs。以下文件未连接 PostgreSQL 实际执行，SQL 行为、查询计划与并发保证仍需在专用练习数据库验证：transactions.mjs、quota-job.mjs。

本章官方参考#

node-postgres Transactions：同一连接和事务命令。
PostgreSQL Transaction Isolation：并发可见性与序列化失败。
PostgreSQL INSERT：ON CONFLICT 与 RETURNING。
node-postgres Pooling：连接获取、归还和池管理。