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(); } };