fix: timeout prisma, mempool race conditions & freezing of tx processing upon erroring

constants/index.ts

@@ -166,7 +166,7 @@ export const HUMAN_READABLE_DATE_FORMAT = 'YYYY-MM-DD'
 
 export const PRICE_API_DATE_FORMAT = 'YYYY-MM-DD'
 export const PRICE_API_TIMEOUT = 40 * 1000 // 40 seconds
-export const PRICE_API_MAX_RETRIES = 3
+export const PRICE_API_MAX_RETRIES = 1
 
 export const SYNC_TXS_JOBS_MAX_RETRIES = 3
 export const SYNC_TXS_JOBS_RETRY_DELAY = 2000
@@ -279,9 +279,9 @@ export const MEMPOOL_PROCESS_DELAY = 100
 // On chronik, the max allowed is 200
 export const CHRONIK_FETCH_N_TXS_PER_PAGE = 200
 
-export const INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY = 512
-export const TX_EMIT_BATCH_SIZE = 10_000 // for our generator, not chronik
-export const DB_COMMIT_BATCH_SIZE = 10_000 // tamanho dos lotes para commit no DB
+export const INITIAL_ADDRESS_SYNC_FETCH_CONCURRENTLY = 128
+export const TX_EMIT_BATCH_SIZE = 2_000 // for our generator, not chronik
+export const DB_COMMIT_BATCH_SIZE = 2_000 // tamanho dos lotes para commit no DB
 
 export const TRIGGER_POST_CONCURRENCY = 100
 export const TRIGGER_EMAIL_CONCURRENCY = 100

jobs/initJobs.ts

@@ -7,7 +7,15 @@ import { syncCurrentPricesWorker, syncBlockchainAndPricesWorker, cleanupClientPa
 EventEmitter.defaultMaxListeners = 20
 
 const main = async (): Promise<void> => {
+  // --- force fresh start ---
   const pricesQueue = new Queue('pricesSync', { connection: redisBullMQ })
+  const blockchainQueue = new Queue('blockchainSync', { connection: redisBullMQ })
+  const cleanupQueue = new Queue('clientPaymentCleanup', { connection: redisBullMQ })
+
+  await pricesQueue.obliterate({ force: true })
+  await blockchainQueue.obliterate({ force: true })
+  await cleanupQueue.obliterate({ force: true })
+
   await pricesQueue.add('syncCurrentPrices',
     {},
     {
@@ -21,18 +29,23 @@ const main = async (): Promise<void> => {
 
   await syncCurrentPricesWorker(pricesQueue.name)
 
-  const blockchainQueue = new Queue('blockchainSync', { connection: redisBullMQ })
-  await blockchainQueue.add('syncBlockchainAndPrices', {}, { jobId: 'syncBlockchainAndPrices' })
+  await blockchainQueue.add('syncBlockchainAndPrices',
+    {},
+    {
+      jobId: 'syncBlockchainAndPrices',
+      removeOnComplete: true,
+      removeOnFail: true
+    }
+  )
   await syncBlockchainAndPricesWorker(blockchainQueue.name)
 
-  const cleanupQueue = new Queue('clientPaymentCleanup', { connection: redisBullMQ })
-
   await cleanupQueue.add(
     'cleanupClientPayments',
     {},
     {
       jobId: 'cleanupClientPayments',
-      removeOnFail: false,
+      removeOnComplete: true,
+      removeOnFail: true,
       repeat: {
         every: CLIENT_PAYMENT_EXPIRATION_TIME
       }

jobs/workers.ts

@@ -56,9 +56,12 @@ export const syncBlockchainAndPricesWorker = async (queueName: string): Promise<
   })
 
   worker.on('failed', (job, err) => {
-    if (job != null) {
-      console.error(`job ${job.id as string}: FAILED — ${err.message}`)
-    }
+    void (async () => {
+      if (job != null) {
+        console.error(`job ${job.id as string}: FAILED — ${err.message}`)
+      }
+      await multiBlockchainClient.destroy()
+    })()
   })
 }
 

package.json

@@ -72,7 +72,7 @@
     "xecaddrjs": "^0.0.1"
   },
   "devDependencies": {
-    "@types/jest": "^27.5.1",
+    "@types/jest": "^30.0.0",
     "@types/node": "^12.0.12",
     "@types/nodemailer": "^6.4.16",
     "@types/react": "^16.9.44",

prisma-local/seeds/prices.ts

@@ -33,7 +33,7 @@ export async function createPricesFile (): Promise<void> {
   const prices: PriceFileData[] = []
 
   await Promise.all(Object.values(NETWORK_TICKERS).map(async (networkTicker) => {
-    const pricesByNetworkTicker = await getAllPricesByNetworkTicker(networkTicker)
+    const pricesByNetworkTicker = await getAllPricesByNetworkTicker(networkTicker, true)
 
     pricesByNetworkTicker?.forEach(price => {
       if (isEmpty(price.Price_in_CAD) || isEmpty(price.Price_in_USD)) { throw new Error(`Price came back with at least one quote empty from API: ${JSON.stringify(price)}`) }

prisma-local/seeds/productionTxs.csv

@@ -131466,4 +131466,8 @@ fe95acd641f4ced15eb81dbfeb508124c322f8ca6cbb8e1868ea6024fcba3574,1000119.19,1754
 baba23425aa5e5bb5c3b2b1b3e9a7f30dd4fe333f58be276700be85ff2cf49f0,1000023.22,1754861586,d1266537-9c17-4ec5-b09f-b05a81811c60,true
 39c84db94eb3c7d9e9b9307060a840ba36dbefeb4add2713f81f911d7bcc3a3f,1000152.16,1754861467,d1266537-9c17-4ec5-b09f-b05a81811c60,truefb2dce1149b2a9a1722954c931964a061704a06910682e6ddd6f676e1c01e0af,1000043.31,1754947609,d1266537-9c17-4ec5-b09f-b05a81811c60,true
 2db424bd1ae48962d0a2eea6bdf6d21fa02af129e7d54d8ecda02511d7d3760a,1000131.92,1754946651,d1266537-9c17-4ec5-b09f-b05a81811c60,true
-eedd73ccf42e75872329a68f2cc0a98513bf58ad0fe8c2314c5a7ea5bb3430a3,1000035.45,1754946080,d1266537-9c17-4ec5-b09f-b05a81811c60,true
+eedd73ccf42e75872329a68f2cc0a98513bf58ad0fe8c2314c5a7ea5bb3430a3,1000035.45,1754946080,d1266537-9c17-4ec5-b09f-b05a81811c60,true924125cf4bb5cbe292d7a1f9a78ea2b03daeac23af8d5b51e8f122ca6dc21ebb,1000030.75,1761165795,d1266537-9c17-4ec5-b09f-b05a81811c60,true
+298aa89e9d0e352492386d4e35f1f53135e6e79d6abdd2043cfcb73ef4e73fec,1000000,1761165132,d1266537-9c17-4ec5-b09f-b05a81811c60,true
+746ad0dde27297c0c29f0e844b488fd06abdc5091e678a2a0e1b0b5c1d5efbd1,1000036.53,1761165079,d1266537-9c17-4ec5-b09f-b05a81811c60,true
+d0180fa8ffb28760f64bdadb2fd7289196f003cd27a2adcee6e6419e848c1597,1000000,1761164173,d1266537-9c17-4ec5-b09f-b05a81811c60,true
+f9f64587e7f7b11f771e130455b59fd84e5a3422215ac1f08505418fd395dc07,1000068.28,1761164139,d1266537-9c17-4ec5-b09f-b05a81811c60,true7e6312169483006fec8d6d1d9d4311c82f3334be4c58fe4a188cbc9e088de75c,1000716.35,1761169691,d1266537-9c17-4ec5-b09f-b05a81811c60,true

scripts/updateAllPriceConnections.ts

@@ -58,7 +58,8 @@ async function fixMisconnectedTxs (): Promise<void> {
           select: {
             price: {
               select: {
-                timestamp: true
+                timestamp: true,
+                networkId: true
               }
             }
           }