hyb-oa/scripts/qa/budget-pool-db-fixture.cjs

const assert = require('node:assert/strict');
const { randomUUID } = require('node:crypto');
const { createRequire } = require('node:module');
const path = require('node:path');
const { setTimeout: delay } = require('node:timers/promises');
const req = createRequire(path.resolve(__dirname, '../../server/package.json'));
const { PrismaClient } = req('@prisma/client');
const { PrismaPg } = req('@prisma/adapter-pg');
const { reserveBudgetForEntity, releaseReservedBudgetIfAny } = req('./src/services/purchaseBudget');
const { authorizeBudgetForEntity, reverseBudgetForEntity } = req('./src/services/travelPayments');
const { lockBudgetPoolsForUpdate, resolveApplicableBudgetPoolByContext } = req('./src/services/budgetPool');

module.exports = async function runBudgetFixture(pool) {
  const prisma = new PrismaClient({ adapter: new PrismaPg(pool, { disposeExternalPool: false }) });
  const passed = []; const failures = [];
  const tx = fn => prisma.$transaction(fn);
  const actorId = randomUUID();
  const createPool = (used = '0.00') => prisma.budgetPool.create({ data: {
    code: `SYN-${randomUUID()}`, name: '合成预算池', period_start: new Date('2026-01-01'),
    period_end: new Date('2026-12-31'), allocated_amount: '1000.00', used_amount: used,
  } });
  const input = (id, amount = '60.00') => ({ budgetPoolId: id, entityType: 'purchase_request',
    entityId: randomUUID(), amount, actorId, note: 'synthetic concurrent budget check' });
  async function check(name, fn) {
    try { await fn(); passed.push(name); console.log(JSON.stringify({ check: name, status: 'PASS' })); }
    catch (error) { failures.push(name); console.log(JSON.stringify({ check: name, status: 'FAIL', code: error.code ?? error.name,
      fixtureLine: /budget-pool-db-fixture.cjs:(\d+)/.exec(error.stack ?? '')?.[1],
      reason: /transaction client/.test(error.message) ? 'transaction_client_required' : undefined,
      causeName: error.cause?.name, causeCode: error.cause?.code })); }
  }
  // Both calls must reach a real database lock wait before release. The old
  // implementation waits at UPDATE after reading stale usage; the fixed one
  // waits before its authoritative read. No simulated locks or timing guess.
  async function contend(id, calls, standalone = false, beforeUnlock) {
    const blocker = await pool.connect(); let results;
    try {
      await blocker.query('BEGIN');
      await blocker.query('SELECT id FROM budget_pools WHERE id=$1 FOR NO KEY UPDATE', [id]);
      const jobs = calls.map(fn => standalone ? fn() : tx(fn));
      let completed;
      results = Promise.allSettled(jobs).then(value => { completed = value; return value; });
      const deadline = Date.now() + 2500;
      let waiting = 0;
      while (Date.now() < deadline) {
        const r = await pool.query("SELECT count(*)::int n FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE '%budget_pools%'");
        waiting = r.rows[0].n;
        if (completed?.some(r => r.status === 'rejected')) throw completed.find(r => r.status === 'rejected').reason;
        if (waiting >= calls.length) break;
        await delay(10);
      }
      assert.equal(waiting, calls.length, 'all contenders reached database lock wait');
      if (beforeUnlock) await beforeUnlock(blocker);
      await blocker.query('COMMIT');
      const settled = await results;
      for (const r of settled) if (r.status === 'rejected') throw r.reason;
      return settled.map(r => r.value);
    } finally {
      await blocker.query('ROLLBACK'); blocker.release();
      if (results) await results;
    }
  }
  const used = async id => (await prisma.budgetPool.findUniqueOrThrow({ where: { id } })).used_amount.toFixed(2);
  try {
    await check('reserve_same_pool_accumulates_both_commits', async () => {
      const p = await createPool();
      await contend(p.id, [t => reserveBudgetForEntity(t, input(p.id)), t => reserveBudgetForEntity(t, input(p.id))]);
      assert.equal(await used(p.id), '120.00');
      assert.equal(await prisma.budgetLedgerEntry.count({ where: { budget_pool_id: p.id } }), 2);
    });
    await check('purchase_and_card_authorization_share_pool', async () => {
      const p = await createPool();
      await contend(p.id, [t => reserveBudgetForEntity(t, input(p.id)), t => authorizeBudgetForEntity(t, input(p.id))]);
      assert.equal(await used(p.id), '120.00');
    });
    await check('release_and_reserve_preserve_both_deltas', async () => {
      const p = await createPool('100.00'); const reserved = input(p.id, '40.00');
      await prisma.budgetLedgerEntry.create({ data: { budget_pool_id: p.id, entity_type: reserved.entityType,
        entity_id: reserved.entityId, kind: 'reserve', amount: reserved.amount } });
      await contend(p.id, [t => releaseReservedBudgetIfAny(t, reserved), t => reserveBudgetForEntity(t, input(p.id, '20.00'))]);
      assert.equal(await used(p.id), '80.00');
    });
    await check('card_reverse_and_reserve_preserve_both_deltas', async () => {
      const p = await createPool('100.00');
      await contend(p.id, [t => reverseBudgetForEntity(t, input(p.id, '40.00')), t => reserveBudgetForEntity(t, input(p.id, '20.00'))]);
      assert.equal(await used(p.id), '80.00');
    });
    await check('ledger_and_usage_rollback_together', async () => {
      const p = await createPool();
      await assert.rejects(tx(async t => { await reserveBudgetForEntity(t, input(p.id)); throw new Error('synthetic rollback'); }), /synthetic rollback/);
      assert.equal(await used(p.id), '0.00');
      assert.equal(await prisma.budgetLedgerEntry.count({ where: { budget_pool_id: p.id } }), 0);
    });
    await check('concurrent_duplicate_reserve_rechecks_ledger_after_wait', async () => {
      const p = await createPool(); const args = input(p.id);
      const call = async t => { try { return await reserveBudgetForEntity(t, args); }
        catch (error) { assert.equal(error.details?.reasonCode, 'purchase_budget.reserve_already_exists'); return 'duplicate'; } };
      const results = await contend(p.id, [call, call]);
      assert.equal(results.filter(r => r === 'duplicate').length, 1);
      assert.equal(await used(p.id), '60.00');
      assert.equal(await prisma.budgetLedgerEntry.count({ where: { budget_pool_id: p.id } }), 1);
    });
    await check('release_multi_pool_reverse_input_order_has_no_deadlock', async () => {
      const a = await createPool('80.00'); const b = await createPool('80.00');
      const left = input(a.id, '40.00'); const right = input(b.id, '40.00');
      for (const [entity, pools] of [[left, [a, b]], [right, [b, a]]]) {
        for (const p of pools) await prisma.budgetLedgerEntry.create({ data: {
          budget_pool_id: p.id, entity_type: entity.entityType, entity_id: entity.entityId, kind: 'reserve', amount: '40.00',
        } });
      }
      await contend([a.id, b.id].sort()[0], [t => releaseReservedBudgetIfAny(t, left), t => releaseReservedBudgetIfAny(t, right)]);
      assert.equal(await used(a.id), '0.00'); assert.equal(await used(b.id), '0.00');
      const second = await tx(t => releaseReservedBudgetIfAny(t, left)); assert.equal(second.applied, false);
    });
    await check('release_retains_nonnegative_floor', async () => {
      const p = await createPool('20.00'); const args = input(p.id, '30.00');
      await prisma.budgetLedgerEntry.create({ data: { budget_pool_id: p.id, entity_type: args.entityType,
        entity_id: args.entityId, kind: 'reserve', amount: args.amount } });
      await tx(t => releaseReservedBudgetIfAny(t, args)); assert.equal(await used(p.id), '0.00');
    });
    await check('different_pools_do_not_share_a_global_lock', async () => {
      const a = await createPool(); const b = await createPool();
      await tx(async t => {
        await lockBudgetPoolsForUpdate(t, [a.id]);
        await tx(other => reserveBudgetForEntity(other, input(b.id)));
      });
      assert.equal(await used(b.id), '60.00');
    });
    await check('candidate_archived_while_waiting_is_not_silently_replaced', async () => {
      const department = await prisma.department.create({ data: { name: '合成候选变更' } });
      const p = await createPool(); await prisma.budgetPool.update({ where: { id: p.id }, data: { department_id: department.id } });
      const blocker = await pool.connect(); let pending;
      try {
        await blocker.query('BEGIN'); await blocker.query('SELECT id FROM budget_pools WHERE id=$1 FOR NO KEY UPDATE', [p.id]);
        pending = tx(t => resolveApplicableBudgetPoolByContext(t, { departmentId: department.id, effectiveDate: new Date('2026-09-19') }, { forUpdate: true }))
          .then(value => ({ value }), error => ({ error }));
        const deadline = Date.now() + 2000; let waiting = 0;
        while (Date.now() < deadline) {
          waiting = (await pool.query("SELECT count(*)::int n FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'")).rows[0].n;
          if (waiting) break; await delay(10);
        }
        assert.equal(waiting, 1);
        await blocker.query("UPDATE budget_pools SET status='archived',archived_at=now() WHERE id=$1", [p.id]);
        await blocker.query('COMMIT');
        assert.equal((await pending).error?.details?.reasonCode, 'budget.pool_changed');
      } finally { await blocker.query('ROLLBACK'); blocker.release(); if (pending) await pending; }
    });
    await require('./budget-pool-http-fixture.cjs')({ prisma, pool, check, contend, used });
    console.log(JSON.stringify({ scope: 'real-services-real-isolated-postgresql', passed, failures }));
    assert.equal(failures.length, 0, 'budget concurrency checks failed');
  } finally { await prisma.$disconnect(); }
};