From dd181b38e84b9e2c869ef7ed706c1a42e07ec62b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 7 Nov 2026 16:49:25 +0000 Subject: [PATCH] fix: keep bulk tx persist interactive transactions short Classify existing rows and apply status updates outside the interactive transaction, group identical updates into updateMany, and lower the create-path timeout from 45s to 15s so pool connections are released sooner. Co-authored-by: David --- constants/index.ts | 2 +- services/transactionService.ts | 180 ++++++++++++++++++--------------- 2 files changed, 102 insertions(+), 80 deletions(-) diff --git a/constants/index.ts b/constants/index.ts index a471e3fa..7729afa7 100644 --- a/constants/index.ts +++ b/constants/index.ts @@ -198,7 +198,7 @@ export const QUOTE_IDS: KeyValueT = { USD: 1, CAD: 2 } export type BLOCKCHAIN_CLIENT_OPTIONS = 'chronik' -export const UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT = 45000 +export const UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT = 15000 export const DEFAULT_TX_PAGE_SIZE = 100 export const PAYMENT_WEEK_KEY_FORMAT = 'YYYY:WW' diff --git a/services/transactionService.ts b/services/transactionService.ts index c503a183..cc219cfd 100644 --- a/services/transactionService.ts +++ b/services/transactionService.ts @@ -681,6 +681,8 @@ export async function filterRowsNeedingCreateMany ( /** * Insert or update transactions and inputs only (no prices, cache, or heavy includes). + * Existing-row lookup and updates stay outside the interactive transaction so a + * pool connection is not held for the full classify + per-row update window. */ async function persistManyTransactionRows ( transactionsData: Prisma.TransactionUncheckedCreateInput[] @@ -705,68 +707,68 @@ async function persistManyTransactionRows ( } }) - const inserted: SyncPersistedTransaction[] = [] - const updated: SyncPersistedTransaction[] = [] - let updatedCount = 0 + const existingTxs = await prisma.transaction.findMany({ + where: { + OR: flatTxData.map(row => ({ + hash: row.hash, + addressId: row.addressId + })) + }, + select: { + id: true, + hash: true, + addressId: true, + confirmed: true, + timestamp: true, + orphaned: true + } + }) - await prisma.$transaction( - async (tx) => { - const existingTxs = await tx.transaction.findMany({ - where: { - OR: flatTxData.map(row => ({ - hash: row.hash, - addressId: row.addressId - })) - }, - select: { - id: true, - hash: true, - addressId: true, - confirmed: true, - timestamp: true, - orphaned: true - } - }) + const existingMap = new Map() + for (const row of existingTxs) { + existingMap.set(`${row.hash}:${row.addressId}`, row) + } - const existingMap = new Map() - for (const row of existingTxs) { - existingMap.set(`${row.hash}:${row.addressId}`, row) + const newTxs: typeof flatTxData = [] + const newTxsInputs: typeof txInputs = [] + const toUpdate: Array<{ + id: string + confirmed: boolean + timestamp: number + orphaned: boolean + }> = [] + + for (let i = 0; i < flatTxData.length; i++) { + const row = flatTxData[i] + const key = `${row.hash}:${row.addressId}` + const existing = existingMap.get(key) + + if (existing != null) { + const confirmedChanged = existing.confirmed !== row.confirmed + const timestampChanged = existing.timestamp !== row.timestamp + const orphanedChanged = existing.orphaned !== row.orphaned + + if (confirmedChanged || timestampChanged || orphanedChanged) { + toUpdate.push({ + id: existing.id, + confirmed: row.confirmed, + timestamp: row.timestamp, + orphaned: row.orphaned + }) } + } else { + newTxs.push(row) + newTxsInputs.push(txInputs[i]) + } + } - const newTxs: typeof flatTxData = [] - const newTxsInputs: typeof txInputs = [] - const toUpdate: Array<{ - id: string - confirmed: boolean - timestamp: number - orphaned: boolean - }> = [] - - for (let i = 0; i < flatTxData.length; i++) { - const row = flatTxData[i] - const key = `${row.hash}:${row.addressId}` - const existing = existingMap.get(key) - - if (existing != null) { - const confirmedChanged = existing.confirmed !== row.confirmed - const timestampChanged = existing.timestamp !== row.timestamp - const orphanedChanged = existing.orphaned !== row.orphaned - - if (confirmedChanged || timestampChanged || orphanedChanged) { - toUpdate.push({ - id: existing.id, - confirmed: row.confirmed, - timestamp: row.timestamp, - orphaned: row.orphaned - }) - } - } else { - newTxs.push(row) - newTxsInputs.push(txInputs[i]) - } - } + const inserted: SyncPersistedTransaction[] = [] + const updated: SyncPersistedTransaction[] = [] + let updatedCount = 0 - if (newTxs.length > 0) { + if (newTxs.length > 0) { + await prisma.$transaction( + async (tx) => { await tx.transaction.createMany({ data: newTxs, skipDuplicates: true @@ -825,33 +827,53 @@ async function persistManyTransactionRows ( for (const createdTx of createdTxs) { inserted.push(createdTx) } + }, + { + timeout: UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT } + ) + } - if (toUpdate.length > 0) { - await Promise.all( - toUpdate.map(async update => - await tx.transaction.update({ - where: { id: update.id }, - data: { - confirmed: update.confirmed, - timestamp: update.timestamp, - orphaned: update.orphaned - } - }) - ) - ) - const updatedTxs = await tx.transaction.findMany({ - where: { id: { in: toUpdate.map(u => u.id) } }, - select: syncPersistedTxSelect + if (toUpdate.length > 0) { + const updateGroups = new Map() + for (const update of toUpdate) { + const key = `${String(update.confirmed)}:${update.timestamp}:${String(update.orphaned)}` + const group = updateGroups.get(key) + if (group == null) { + updateGroups.set(key, { + confirmed: update.confirmed, + timestamp: update.timestamp, + orphaned: update.orphaned, + ids: [update.id] }) - updated.push(...updatedTxs) - updatedCount = updatedTxs.length + } else { + group.ids.push(update.id) } - }, - { - timeout: UPSERT_TRANSACTION_PRICES_ON_DB_TIMEOUT } - ) + + for (const group of updateGroups.values()) { + await prisma.transaction.updateMany({ + where: { id: { in: group.ids } }, + data: { + confirmed: group.confirmed, + timestamp: group.timestamp, + orphaned: group.orphaned + } + }) + } + + const updatedTxs = await prisma.transaction.findMany({ + where: { id: { in: toUpdate.map(u => u.id) } }, + select: syncPersistedTxSelect + }) + updated.push(...updatedTxs) + updatedCount = updatedTxs.length + } return { inserted, updated, updatedCount } }