Farhan/fix-blockreader (#98)

* trying to handle a weird error

* updated query to fetch txHash

Author: farhanW3

worker/controller/blockchainReader.ts

@@ -1,6 +1,6 @@
-import { ChainOrRpc } from "@thirdweb-dev/sdk";
 import { BigNumber } from "ethers";
 import { FastifyInstance } from "fastify";
+import { Knex } from "knex";
 import { connectWithDatabase, env, getSDK } from "../../core";
 import {
   getSubmittedTransactions,
@@ -12,52 +12,69 @@ const MINED_TX_CRON_ENABLED = env.MINED_TX_CRON_ENABLED;
 export const checkForMinedTransactionsOnBlockchain = async (
   server: FastifyInstance,
 ) => {
-  const knex = await connectWithDatabase();
-  if (!MINED_TX_CRON_ENABLED) {
-    server.log.warn("Mined Tx Cron is disabled");
-    return;
-  }
-  server.log.info("Running Cron to check for mined transactions on blockchain");
-  const trx = await knex.transaction();
-  const transactions = await getSubmittedTransactions(knex, trx);
-  if (transactions.length === 0) {
-    server.log.warn("No transactions to check for mined status");
+  let knex: Knex | undefined;
+  let trx: Knex.Transaction | undefined;
+  try {
+    knex = await connectWithDatabase();
+    if (!MINED_TX_CRON_ENABLED) {
+      server.log.warn("Mined Tx Cron is disabled");
+      return;
+    }
+    server.log.info(
+      "Running Cron to check for mined transactions on blockchain",
+    );
+    trx = await knex.transaction();
+    const transactions = await getSubmittedTransactions(knex);
+    if (transactions.length === 0) {
+      server.log.warn("No transactions to check for mined status");
+      await trx.commit();
+      await trx.destroy();
+      await knex.destroy();
+      return;
+    }
+
+    const txReceipts = await Promise.all(
+      transactions.map(async (txData) => {
+        server.log.debug(
+          `Getting receipt for tx: ${txData.txHash} on chain: ${txData.chainId} for queueId: ${txData.identifier}`,
+        );
+        const sdk = await getSDK(txData.chainId!);
+        return sdk.getProvider().getTransactionReceipt(txData.txHash!);
+      }),
+    );
+
+    for (let txReceipt of txReceipts) {
+      const txData = transactions.find(
+        (tx) => tx.txHash === txReceipt.transactionHash,
+      );
+      if (txData) {
+        server.log.debug(
+          `Got receipt for tx: ${txData.txHash} ${txData.identifier}, ${txReceipt.effectiveGasPrice}`,
+        );
+        await updateTransactionState(
+          knex,
+          txData.identifier!,
+          "mined",
+          trx,
+          undefined,
+          undefined,
+          { gasPrice: BigNumber.from(txReceipt.effectiveGasPrice).toString() },
+        );
+      }
+    }
     await trx.commit();
     await trx.destroy();
     await knex.destroy();
     return;
-  }
-
-  const txReceipts = await Promise.all(
-    transactions.map(async (txData) => {
-      server.log.debug(
-        `Getting receipt for tx: ${txData.txHash} on chain: ${txData.chainId} for queueId: ${txData.identifier}`,
-      );
-      const sdk = await getSDK(txData.chainId as ChainOrRpc);
-      return sdk.getProvider().getTransactionReceipt(txData.txHash!);
-    }),
-  );
-
-  for (let txReceipt of txReceipts) {
-    const txData = transactions.find(
-      (tx) => tx.txHash === txReceipt.transactionHash,
-    );
-    if (txData) {
-      server.log.debug(
-        `Got receipt for tx: ${txData.txHash} ${txData.identifier}, ${txReceipt.effectiveGasPrice}`,
-      );
-      await updateTransactionState(
-        knex,
-        txData.identifier!,
-        "mined",
-        trx,
-        undefined,
-        undefined,
-        { gasPrice: BigNumber.from(txReceipt.effectiveGasPrice).toString() },
-      );
+  } catch (error) {
+    if (trx) {
+      await trx.rollback();
+      await trx.destroy();
     }
+    if (knex) {
+      await knex.destroy();
+    }
+    server.log.error(error);
+    return;
   }
-  await trx.commit();
-  await trx.destroy();
-  await knex.destroy();
 };

worker/services/dbOperations.ts

@@ -171,19 +171,16 @@ export const getWalletDetailsWithoutTrx = async (
 
 export const getSubmittedTransactions = async (
   database: Knex,
-  transaction: Knex.Transaction,
 ): Promise<TransactionSchema[]> => {
-  const data = await database
-    .raw(
-      `select * from transactions
+  const data = await database.raw(
+    `select * from transactions
     where "txProcessed" = true
     and "txSubmitted" = true
     and "txMined" = false
     and "txErrored" = false
+    and "txHash" is not null
     order by "txSubmittedTimestamp" ASC
-    limit ${MIN_TX_TO_CHECK_FOR_MINED_STATUS}
-    for update skip locked`,
-    )
-    .transacting(transaction);
+    limit ${MIN_TX_TO_CHECK_FOR_MINED_STATUS}`,
+  );
   return data.rows;
 };