refat: added a table to track blocks that were not saved in the database.

Author: mschead

indexer/src/models/streaming-error.ts

@@ -0,0 +1,36 @@
+import { Model, DataTypes, Optional } from "sequelize";
+import { sequelize } from "../config/database";
+
+interface StreamingErrorAttributes {
+  id: number;
+  chainId: number;
+  hash: string;
+}
+
+interface StreamingErrorCreationAttributes
+  extends Optional<StreamingErrorAttributes, "id"> {}
+
+class StreamingError
+  extends Model<StreamingErrorAttributes, StreamingErrorCreationAttributes>
+  implements StreamingErrorAttributes
+{
+  declare id: number;
+  declare chainId: number;
+  declare hash: string;
+}
+
+StreamingError.init(
+  {
+    id: { type: DataTypes.INTEGER, autoIncrement: true, primaryKey: true },
+    chainId: { type: DataTypes.INTEGER },
+    hash: { type: DataTypes.STRING },
+  },
+  {
+    sequelize,
+    modelName: "StreamingError",
+    tableName: "StreamingErrors",
+    timestamps: true,
+  },
+);
+
+export default StreamingError;

indexer/src/services/sync/guards.ts

@@ -2,7 +2,6 @@ import pLimit from "p-limit";
 import { rootPgPool, sequelize } from "../../config/database";
 import { getGuardsFromBalances } from "./payload";
 import Guard from "../../models/guard";
-import { delay } from "../../utils/helpers";
 
 const CONCURRENCY_LIMIT = 4; // Number of concurrent fetches allowed
 const limitFetch = pLimit(CONCURRENCY_LIMIT);

indexer/src/services/sync/streaming.ts

@@ -5,6 +5,7 @@ import { dispatch, DispatchInfo } from "../../jobs/publisher-job";
 import { uint64ToInt64 } from "../../utils/int-uint-64";
 import Block, { BlockAttributes } from "../../models/block";
 import { sequelize } from "../../config/database";
+import StreamingError from "../../models/streaming-error";
 
 const SYNC_BASE_URL = getRequiredEnvString("SYNC_BASE_URL");
 const SYNC_NETWORK = getRequiredEnvString("SYNC_NETWORK");
@@ -44,26 +45,39 @@ export async function startStreaming() {
       blocksToProcess.push(b);
     }
 
+    console.log("Processing blocks:", blocksToProcess.length);
     const promises = blocksToProcess.map(async (block: any) => {
-      try {
-        return saveBlock(block);
-      } catch (error) {
-        console.error("Error saving block:", error);
+      const blockData = await saveBlock(block);
+      if (blockData === null) {
+        await StreamingError.create({
+          hash: block.header.hash,
+          chainId: block.header.chainId,
+        });
       }
+      return blockData;
     });
 
-    const res = (await Promise.all(promises)).filter(
-      (r) => r !== null,
+    const processed = (await Promise.all(promises)).filter(
+      (r) => r !== null || r !== undefined,
     ) as DispatchInfo[];
 
-    const dispatches = res.map(async (r, index) => {
-      await dispatch(r);
-      await delay(500);
-      console.log("Dispatched block:", index);
+    const dispatches = processed.map(async (r) => {
+      try {
+        await delay(500);
+        await dispatch(r);
+      } catch (err) {
+        console.error("Error dispatching block:", err);
+      }
     });
 
     await Promise.all(dispatches);
-    console.log("Done processing blocks: ", blocksToProcess.length);
+    console.log(
+      "Processed:",
+      processed.length,
+      "|",
+      "Dispatched:",
+      dispatches.length,
+    );
   }, 1000 * 10);
 
   setInterval(