perf: remove paid result account lock contention
This commit is contained in:
hectorzhao
@@ -11,7 +11,7 @@ function createPrismaMock() {
|
||||
findMany: jest.fn().mockResolvedValue([]),
|
||||
},
|
||||
tenantAccount: {
|
||||
findMany: jest.fn(),
|
||||
findMany: jest.fn().mockResolvedValue([]),
|
||||
findUnique: jest.fn().mockImplementation(() => Promise.resolve({ ...accountState })),
|
||||
findUniqueOrThrow: jest.fn().mockImplementation(() => Promise.resolve({ ...accountState })),
|
||||
create: jest.fn(),
|
||||
@@ -30,7 +30,7 @@ function createPrismaMock() {
|
||||
}),
|
||||
},
|
||||
accountTransaction: {
|
||||
findMany: jest.fn(),
|
||||
findMany: jest.fn().mockResolvedValue([]),
|
||||
findFirst: jest.fn(),
|
||||
findUnique: jest.fn().mockResolvedValue(null),
|
||||
findUniqueOrThrow: jest.fn().mockImplementation(({ where }) => Promise.resolve({ id: 'tx-charged', idempotencyKey: where.idempotencyKey })),
|
||||
@@ -55,6 +55,7 @@ function createPrismaMock() {
|
||||
create: jest.fn().mockResolvedValue({ id: 'operation-1' }),
|
||||
},
|
||||
$executeRaw: jest.fn(),
|
||||
$queryRaw: jest.fn(),
|
||||
};
|
||||
return Object.assign(prisma, {
|
||||
$transaction: jest.fn((callback: (client: typeof prisma) => unknown) => callback(prisma)),
|
||||
@@ -319,11 +320,11 @@ describe('BillingService', () => {
|
||||
it('settles a frozen SMS charge with one account lock and an idempotent ledger pair', async () => {
|
||||
const prisma = createPrismaMock();
|
||||
const rows = new Map<string, Record<string, unknown>>();
|
||||
prisma.accountTransaction.findUnique.mockImplementation(({ where }) => Promise.resolve(rows.get(where.idempotencyKey) ?? null));
|
||||
prisma.accountTransaction.findUniqueOrThrow.mockImplementation(({ where }) => Promise.resolve(rows.get(where.idempotencyKey)));
|
||||
prisma.accountTransaction.createMany.mockImplementation(({ data }) => {
|
||||
data.forEach((row: Record<string, unknown>) => rows.set(String(row.idempotencyKey), { id: `tx-${row.transactionType}`, ...row }));
|
||||
return Promise.resolve({ count: data.length });
|
||||
prisma.accountTransaction.findMany.mockImplementation(() => Promise.resolve([...rows.values()]));
|
||||
prisma.$queryRaw.mockImplementation(() => {
|
||||
const charged = { id: 'tx-charged', idempotencyKey: 'sms-charge:msg-paid-1', transactionType: 'charged', amountCents: -325 };
|
||||
rows.set(String(charged.idempotencyKey), charged);
|
||||
return Promise.resolve([charged]);
|
||||
});
|
||||
const service = new BillingService(prisma as never);
|
||||
const input = { tenantId: 'tenant-1', amountCents: 325, messageId: 'msg-paid-1', taskId: 'task-paid-1' };
|
||||
@@ -333,7 +334,8 @@ describe('BillingService', () => {
|
||||
|
||||
expect(first).toEqual(expect.objectContaining({ transactionType: 'charged', amountCents: -325 }));
|
||||
expect(replay).toEqual(first);
|
||||
expect(prisma.accountTransaction.createMany).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.$queryRaw).toHaveBeenCalledTimes(1);
|
||||
expect(prisma.$executeRaw).not.toHaveBeenCalled();
|
||||
expect(prisma.tenantAccount.update).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
|
||||
@@ -441,44 +441,49 @@ export class BillingService {
|
||||
if (amountCents <= 0) return null;
|
||||
const releaseKey = `sms-charge-release:${data.messageId}`;
|
||||
const chargeKey = `sms-charge:${data.messageId}`;
|
||||
return this.prisma.$transaction(async (tx) => {
|
||||
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'tenant-account:' + data.tenantId}, 0))`;
|
||||
const [release, charge] = await Promise.all([
|
||||
tx.accountTransaction.findUnique({ where: { idempotencyKey: releaseKey } }),
|
||||
tx.accountTransaction.findUnique({ where: { idempotencyKey: chargeKey } }),
|
||||
]);
|
||||
if (charge) return charge;
|
||||
const account = await tx.tenantAccount.findUniqueOrThrow({ where: { tenantId: data.tenantId } });
|
||||
const balance = moneyToNumber(account.balanceCents);
|
||||
if (release) {
|
||||
const updated = await tx.tenantAccount.update({
|
||||
where: { tenantId: data.tenantId },
|
||||
data: { balanceCents: { decrement: amountCents } },
|
||||
});
|
||||
return tx.accountTransaction.create({
|
||||
data: {
|
||||
tenantId: data.tenantId, transactionType: 'charged', idempotencyKey: chargeKey,
|
||||
amountCents: -amountCents, balanceAfter: updated.balanceCents,
|
||||
relatedType: 'sms_message_record', relatedId: data.messageId, remark: '提交成功扣费',
|
||||
},
|
||||
});
|
||||
}
|
||||
await tx.accountTransaction.createMany({
|
||||
data: [
|
||||
{
|
||||
tenantId: data.tenantId, transactionType: 'released', idempotencyKey: releaseKey,
|
||||
amountCents, balanceAfter: balance + amountCents,
|
||||
relatedType: 'sms_batch_task', relatedId: data.taskId, remark: data.remark,
|
||||
},
|
||||
{
|
||||
tenantId: data.tenantId, transactionType: 'charged', idempotencyKey: chargeKey,
|
||||
amountCents: -amountCents, balanceAfter: balance,
|
||||
relatedType: 'sms_message_record', relatedId: data.messageId, remark: '提交成功扣费',
|
||||
},
|
||||
],
|
||||
const existing = await this.prisma.accountTransaction.findMany({
|
||||
where: { idempotencyKey: { in: [releaseKey, chargeKey] } },
|
||||
});
|
||||
const charge = existing.find((row) => row.idempotencyKey === chargeKey);
|
||||
if (charge) return charge;
|
||||
const release = existing.find((row) => row.idempotencyKey === releaseKey);
|
||||
if (release) {
|
||||
// Recover the legacy two-transaction boundary: a crash may have committed
|
||||
// release before charge, so this path must perform the missing balance debit.
|
||||
return this.applyAccountDelta({
|
||||
tenantId: data.tenantId, transactionType: 'charged', idempotencyKey: chargeKey,
|
||||
amountCents: -amountCents, relatedType: 'sms_message_record', relatedId: data.messageId,
|
||||
remark: '提交成功扣费（恢复既有已释放冻结）',
|
||||
});
|
||||
return tx.accountTransaction.findUniqueOrThrow({ where: { idempotencyKey: chargeKey } });
|
||||
}, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted });
|
||||
}
|
||||
const rows = await this.prisma.$queryRaw<Array<{ id: string; idempotencyKey: string }>>(Prisma.sql`
|
||||
WITH account AS (
|
||||
SELECT "balanceCents" FROM "TenantAccount" WHERE "tenantId" = ${data.tenantId}
|
||||
), inserted AS (
|
||||
INSERT INTO "AccountTransaction" (
|
||||
id, "tenantId", "transactionType", "idempotencyKey", "amountCents",
|
||||
"balanceAfter", "relatedType", "relatedId", remark, "createdAt"
|
||||
)
|
||||
SELECT gen_random_uuid()::text, ${data.tenantId}, ledger."transactionType", ledger."idempotencyKey",
|
||||
ledger."amountCents", account."balanceCents" + ledger."balanceDelta",
|
||||
ledger."relatedType", ledger."relatedId", ledger.remark, (NOW() AT TIME ZONE 'UTC')
|
||||
FROM account
|
||||
CROSS JOIN (VALUES
|
||||
('released', ${releaseKey}, ${amountCents}::bigint, ${amountCents}::bigint, 'sms_batch_task', ${data.taskId}, ${data.remark ?? null}),
|
||||
('charged', ${chargeKey}, ${-amountCents}::bigint, 0::bigint, 'sms_message_record', ${data.messageId}, '提交成功扣费')
|
||||
) AS ledger("transactionType", "idempotencyKey", "amountCents", "balanceDelta", "relatedType", "relatedId", remark)
|
||||
ON CONFLICT ("idempotencyKey") DO NOTHING
|
||||
RETURNING id, "idempotencyKey"
|
||||
)
|
||||
SELECT id, "idempotencyKey" FROM inserted WHERE "idempotencyKey" = ${chargeKey}
|
||||
UNION ALL
|
||||
SELECT id, "idempotencyKey" FROM "AccountTransaction" WHERE "idempotencyKey" = ${chargeKey}
|
||||
LIMIT 1
|
||||
`);
|
||||
if (rows[0]) return rows[0];
|
||||
// A concurrent identical callback can win ON CONFLICT while remaining
|
||||
// invisible to this statement's snapshot; one read repairs that MVCC edge.
|
||||
return this.prisma.accountTransaction.findUniqueOrThrow({ where: { idempotencyKey: chargeKey } });
|
||||
}
|
||||
|
||||
release(data: BillingActionDto) {
|
||||
|
||||
Reference in New Issue
Block a user