use transaction (#731)

* use transaction * add fault saftey to blockTime and check for finalized
helium · Nov 1, 2024 · fa4687c · fa4687c
1 parent c447ed8
commit fa4687c
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 19 deletions.
diff --git a/packages/account-postgres-sink-service/src/utils/getBlockTimeWithRetry.ts b/packages/account-postgres-sink-service/src/utils/getBlockTimeWithRetry.ts
@@ -1,13 +1,15 @@
-import * as anchor from '@coral-xyz/anchor';
+import * as anchor from "@coral-xyz/anchor";
 
 export const getBlockTimeWithRetry = async ({
   slot,
   maxRetries = 3,
+  maxSlotIncrement = 1,
   retryInterval = 1000,
   provider,
 }: {
   slot: number;
   maxRetries?: number;
+  maxSlotIncrement?: number;
   retryInterval?: number;
   provider: anchor.AnchorProvider;
 }): Promise<number | null> => {
@@ -18,8 +20,8 @@ export const getBlockTimeWithRetry = async ({
     const blockTime = await connection.getBlockTime(slot);
 
     return blockTime;
-  } catch (error) {
-    console.error('Error fetching block time:', error);
+  } catch (err) {
+    console.error("Error fetching block time:", err);
 
     if (maxRetries > 0) {
       console.log(`Retrying in ${retryInterval / 1000} seconds...`);
@@ -31,7 +33,21 @@ export const getBlockTimeWithRetry = async ({
         provider,
       });
     } else {
-      throw new Error('Max retries reached. Unable to fetch block time.');
+      if (
+        maxSlotIncrement > 0 &&
+        (err as Error).message.toLowerCase().includes("slot") &&
+        (err as Error).message.toLowerCase().includes("was skipped")
+      ) {
+        return getBlockTimeWithRetry({
+          slot: slot + 1,
+          maxRetries: 1,
+          maxSlotIncrement: maxSlotIncrement - 1,
+          retryInterval,
+          provider,
+        });
+      }
+
+      throw new Error("Max retries reached. Unable to fetch block time.");
     }
   }
 };
diff --git a/packages/account-postgres-sink-service/src/utils/getTransactionSignaturesUpToBlock.ts b/packages/account-postgres-sink-service/src/utils/getTransactionSignaturesUpToBlock.ts
@@ -1,5 +1,5 @@
-import * as anchor from '@coral-xyz/anchor';
-import { PublicKey, TransactionSignature } from '@solana/web3.js';
+import * as anchor from "@coral-xyz/anchor";
+import { PublicKey, TransactionSignature } from "@solana/web3.js";
 
 interface GetTransactionSignaturesUptoBlockTimeArgs {
   programId: PublicKey;
@@ -26,7 +26,7 @@ export const getTransactionSignaturesUptoBlockTime = async ({
       {
         before: beforeSignature,
       },
-      'confirmed'
+      "finalized"
     );
 
     if (
@@ -54,7 +54,7 @@ export const getTransactionSignaturesUptoBlockTime = async ({
       provider,
     });
   } catch (err) {
-    console.error('Error fetching transaction signatures:', err);
+    console.error("Error fetching transaction signatures:", err);
     return [];
   }
 };
diff --git a/packages/account-postgres-sink-service/src/utils/integrityCheckProgramAccounts.ts b/packages/account-postgres-sink-service/src/utils/integrityCheckProgramAccounts.ts
@@ -61,7 +61,6 @@ export const integrityCheckProgramAccounts = async ({
     const t = await sequelize.transaction({
       isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED,
     });
-    const now = new Date().toISOString();
     const txIdsByAccountId: { [key: string]: string[] } = {};
     const corrections: {
       type: string;
@@ -98,7 +97,7 @@ export const integrityCheckProgramAccounts = async ({
             retry(
               () =>
                 connection.getParsedTransactions(chunk, {
-                  commitment: "confirmed",
+                  commitment: "finalized",
                   maxSupportedTransactionVersion: 0,
                 }),
               retryOptions
@@ -178,9 +177,12 @@ export const integrityCheckProgramAccounts = async ({
             if (accName) {
               const omitKeys = ["refreshed_at", "createdAt"];
               const model = sequelize.models[accName];
-              const existing = await model.findByPk(c.pubkey);
+              const existing = await model.findByPk(c.pubkey, {
+                transaction: t,
+              });
+
               let sanitized = {
-                refreshed_at: now,
+                refreshed_at: new Date().toISOString(),
                 address: c.pubkey,
                 ...sanitizeAccount(decodedAcc),
               };
@@ -191,14 +193,12 @@ export const integrityCheckProgramAccounts = async ({
                 }
               }
 
-              const isEqual =
-                existing &&
-                deepEqual(
-                  _omit(sanitized, omitKeys),
-                  _omit(existing.dataValues, omitKeys)
-                );
+              const shouldUpdate = !deepEqual(
+                _omit(sanitized, omitKeys),
+                _omit(existing?.dataValues, omitKeys)
+              );
 
-              if (!isEqual) {
+              if (shouldUpdate) {
                 corrections.push({
                   type: accName,
                   accountId: c.pubkey,