fix(billing): drop transaction wrapper in recordUsage to relieve pool contention (#4494)

* fix(billing): drop transaction wrapper in recordUsage to relieve pool contention

* fix(billing): always warn on userStats drift in recordUsage
This commit is contained in:
Theodore Li
2026-05-07 13:33:42 -04:00
committed by GitHub
parent 49713f887f
commit ec5793fbaf
+47 -36
View File
@@ -71,15 +71,17 @@ export interface RecordUsageParams {
}
/**
* Records usage in a single atomic transaction.
* Records usage by inserting into usage_log and incrementing userStats counters.
*
* Inserts all entries into usage_log and updates userStats counters
* (totalCost, currentPeriodCost, lastActive) within one Postgres transaction.
* The total cost added to userStats is derived from summing entry costs,
* ensuring usage_log and currentPeriodCost can never drift apart.
* The two writes are intentionally not wrapped in a transaction: under high
* concurrency for the same userId, holding BEGIN/COMMIT across the user_stats
* row-lock wait pins pgbouncer connections and exhausts the pool.
*
* If billing is disabled, total cost is zero, or no entries have positive cost,
* this function returns early without writing anything.
* usage_log is the source of truth and the INSERT propagates errors to the
* caller. The userStats UPDATE is best-effort: failures (and missing-row
* cases) are logged as warnings and swallowed. Counter drift is acceptable
* here — the long-term plan is to derive counters from usage_log directly.
* Any drift warning in logs is a signal that needs investigation.
*/
export async function recordUsage(params: RecordUsageParams): Promise<void> {
if (!isBillingEnabled) {
@@ -103,47 +105,56 @@ export async function recordUsage(params: RecordUsageParams): Promise<void> {
? Object.fromEntries(Object.entries(additionalStats).filter(([k]) => !RESERVED_KEYS.has(k)))
: undefined
await db.transaction(async (tx) => {
if (validEntries.length > 0) {
await tx.insert(usageLog).values(
validEntries.map((entry) => ({
id: generateId(),
userId,
category: entry.category,
source: entry.source,
description: entry.description,
metadata: entry.metadata ?? null,
cost: entry.cost.toString(),
workspaceId: workspaceId ?? null,
workflowId: workflowId ?? null,
executionId: executionId ?? null,
}))
)
}
if (validEntries.length > 0) {
await db.insert(usageLog).values(
validEntries.map((entry) => ({
id: generateId(),
userId,
category: entry.category,
source: entry.source,
description: entry.description,
metadata: entry.metadata ?? null,
cost: entry.cost.toString(),
workspaceId: workspaceId ?? null,
workflowId: workflowId ?? null,
executionId: executionId ?? null,
}))
)
}
const updateFields: Record<string, SQL | Date> = {
lastActive: new Date(),
...(totalCost > 0 && {
totalCost: sql`total_cost + ${totalCost}`,
currentPeriodCost: sql`current_period_cost + ${totalCost}`,
}),
...safeStats,
}
const updateFields: Record<string, SQL | Date> = {
lastActive: new Date(),
...(totalCost > 0 && {
totalCost: sql`total_cost + ${totalCost}`,
currentPeriodCost: sql`current_period_cost + ${totalCost}`,
}),
...safeStats,
}
const result = await tx
try {
const result = await db
.update(userStats)
.set(updateFields)
.where(eq(userStats.userId, userId))
.returning({ userId: userStats.userId })
if (result.length === 0) {
logger.warn('recordUsage: userStats row not found, transaction will roll back', {
logger.warn('recordUsage: userStats row not found; counter increment dropped', {
userId,
totalCost,
hadEntries: validEntries.length > 0,
additionalStatsKeys: safeStats ? Object.keys(safeStats) : [],
})
throw new Error(`userStats row not found for userId: ${userId}`)
}
})
} catch (error) {
logger.warn('recordUsage: userStats update failed; counter increment dropped', {
error: toError(error).message,
userId,
totalCost,
hadEntries: validEntries.length > 0,
additionalStatsKeys: safeStats ? Object.keys(safeStats) : [],
})
}
logger.debug('Recorded usage', {
userId,