Merge pull request #31 from 0xPolygon/changes-sync

feat: new changes
0xPolygon · Mar 4, 2024 · dbc73e2 · dbc73e2
2 parents a11fab7 + cf4bb98
commit dbc73e2
Show file tree

Hide file tree

Showing 22 changed files with 165 additions and 141 deletions.
diff --git a/examples/matic_transfer/consumer/src/consumer.ts b/examples/matic_transfer/consumer/src/consumer.ts
@@ -26,10 +26,10 @@ dotenv.config()
 export default async function startConsuming(transferService: TransferService, transferMapper: TransferMapper): Promise<void> {
     try {
         consume({
-            "metadata.broker.list": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
-            "group.id": process.env.CONSUMER_GROUP_ID || "matic.transfer.consumer",
+            "metadata.broker.list": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
+            "group.id": process.env.CONSUMER_GROUP_ID ?? "matic.transfer.consumer",
             "security.protocol": "plaintext",
-            topic: process.env.TRANSFER_TOPIC || "apps.1.matic.transfer",
+            topic: process.env.TRANSFER_TOPIC ?? "apps.1.matic.transfer",
             coders: {
                 fileName: "matic_transfer",
                 packageName: "matictransferpackage",

diff --git a/examples/matic_transfer/consumer/src/index.ts b/examples/matic_transfer/consumer/src/index.ts
@@ -23,7 +23,7 @@ async function start(): Promise<void> {
             }
         });
 
-        const database = new Database(process.env.MONGO_URL || 'mongodb://localhost:27017/chain-indexer');
+        const database = new Database(process.env.MONGO_URL ?? 'mongodb://localhost:27017/chain-indexer');
         await database.connect();
 
         const transferService = new TransferService(

diff --git a/examples/matic_transfer/consumer/src/interfaces/transfer_methods.ts b/examples/matic_transfer/consumer/src/interfaces/transfer_methods.ts
@@ -16,7 +16,7 @@ const statics = {
         //@ts-ignore
         const tx = await this.findOne().sort({ timestamp: -1 }).exec();
 
-        return tx?.blockNumber || 0;
+        return tx?.blockNumber ?? 0;
     },
 
     /**

diff --git a/examples/matic_transfer/producer/src/index.ts b/examples/matic_transfer/producer/src/index.ts
@@ -21,12 +21,12 @@ Logger.create({
 const producer = produce<ErigonBlockProducer>({
     startBlock: parseInt(process.env.START_BLOCK as string),
     rpcWsEndpoints: process.env.RPC_WS_ENDPOINT_URL_LIST?.split(','),
-    topic: process.env.PRODUCER_TOPIC || "polygon.1.blocks",
+    topic: process.env.PRODUCER_TOPIC ?? "polygon.1.blocks",
     maxReOrgDepth: 96,
     maxRetries: 5,
-    mongoUrl: process.env.MONGO_URL || 'mongodb://localhost:27017/chain-indexer',
+    mongoUrl: process.env.MONGO_URL ?? 'mongodb://localhost:27017/chain-indexer',
     blockSubscriptionTimeout: 120000,
-    "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+    "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
     "security.protocol": "plaintext",
     type: "blocks:erigon"
 });

diff --git a/examples/matic_transfer/transformer/src/index.ts b/examples/matic_transfer/transformer/src/index.ts
@@ -25,7 +25,7 @@ Logger.create({
 try {
     startTransforming(
         {
-            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
             "group.id": "matic.transfer.transformer",
             "security.protocol": "plaintext",
             "message.max.bytes": 26214400,
@@ -35,11 +35,11 @@ try {
                 packageName: "blockpackage",
                 messageType: "Block"
             },
-            topic: process.env.CONSUMER_TOPIC || "polygon.1.blocks",
+            topic: process.env.CONSUMER_TOPIC ?? "polygon.1.blocks",
         },
         {
-            topic: process.env.PRODUCER_TOPIC || "apps.1.matic.transfer",
-            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+            topic: process.env.PRODUCER_TOPIC ?? "apps.1.matic.transfer",
+            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
             "security.protocol": "plaintext",
             "message.max.bytes": 26214400,
             coder: {

diff --git a/examples/nft_balancer/consumer/src/consumer.ts b/examples/nft_balancer/consumer/src/consumer.ts
@@ -26,10 +26,10 @@ dotenv.config()
 export default async function startConsuming(transferTokenService: TransferTokenService, transferTokenMapper: TransferTokenMapper): Promise<void> {
     try {
         consume({
-            "metadata.broker.list": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
-            "group.id": process.env.CONSUMER_GROUP_ID || "matic.transfer.consumer",
+            "metadata.broker.list": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
+            "group.id": process.env.CONSUMER_GROUP_ID ?? "matic.transfer.consumer",
             "security.protocol": "plaintext",
-            topic: process.env.TRANSFER_TOPIC || "apps.1.matic.transfer",
+            topic: process.env.TRANSFER_TOPIC ?? "apps.1.matic.transfer",
             coders: {
                 fileName: "nft_transfer",
                 packageName: "nfttransferpackage",

diff --git a/examples/nft_balancer/consumer/src/index.ts b/examples/nft_balancer/consumer/src/index.ts
@@ -23,7 +23,7 @@ async function start(): Promise<void> {
             }
         });
 
-        const database = new Database(process.env.MONGO_URL || 'mongodb://localhost:27017/chain-indexer');
+        const database = new Database(process.env.MONGO_URL ?? 'mongodb://localhost:27017/chain-indexer');
         await database.connect();
 
         const transferService = new TransferTokenService(

diff --git a/examples/nft_balancer/producer/src/index.ts b/examples/nft_balancer/producer/src/index.ts
@@ -22,12 +22,12 @@ const producer = produce<BlockPollerProducer>({
     startBlock: parseInt(process.env.START_BLOCK as string),
     rpcWsEndpoints: process.env.HTTP_PROVIDER ? [process.env.HTTP_PROVIDER] : undefined,
     blockPollingTimeout: parseInt(process.env.BLOCK_POLLING_TIMEOUT as string),
-    topic: process.env.PRODUCER_TOPIC || "polygon.1442.blocks",
+    topic: process.env.PRODUCER_TOPIC ?? "polygon.1442.blocks",
     maxReOrgDepth: 0,
     maxRetries: 5,
-    mongoUrl: process.env.MONGO_URL || 'mongodb://localhost:27017/chain-indexer',
+    mongoUrl: process.env.MONGO_URL ?? 'mongodb://localhost:27017/chain-indexer',
     // blockSubscriptionTimeout: 120000,
-    "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+    "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
     "security.protocol": "plaintext",
     type: "blocks:polling"
 });

diff --git a/examples/nft_balancer/transformer/src/index.ts b/examples/nft_balancer/transformer/src/index.ts
@@ -25,7 +25,7 @@ Logger.create({
 try {
     startTransforming(
         {
-            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
             "group.id": "matic.transfer.transformer",
             "security.protocol": "plaintext",
             "message.max.bytes": 26214400,
@@ -35,11 +35,11 @@ try {
                 packageName: "blockpackage",
                 messageType: "Block"
             },
-            topic: process.env.CONSUMER_TOPIC || "polygon.1.blocks",
+            topic: process.env.CONSUMER_TOPIC ?? "polygon.1.blocks",
         },
         {
-            topic: process.env.PRODUCER_TOPIC || "apps.1.matic.transfer",
-            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL || "localhost:9092",
+            topic: process.env.PRODUCER_TOPIC ?? "apps.1.matic.transfer",
+            "bootstrap.servers": process.env.KAFKA_CONNECTION_URL ?? "localhost:9092",
             "security.protocol": "plaintext",
             "message.max.bytes": 26214400,
             coder: {

diff --git a/internal/block_getters/quicknode_block_getter.ts b/internal/block_getters/quicknode_block_getter.ts
@@ -36,31 +36,42 @@ export class QuickNodeBlockGetter extends BlockGetter implements IBlockGetter {
      */
     public async getBlockWithTransactionReceipts(blockNumber: number | string, retryCount: number = 0): Promise<IBlock> {
         try {
-            const response: IQuickNodeResponse = await new Promise((resolve, reject) => {
-                const timeout = setTimeout(() => {
-                    reject(new Error(`Request timed out for block: ${blockNumber}`));
-                }, this.rpcTimeout || 4000);
+            const response: IQuickNodeResponse = await new Promise(
+                async (resolve, reject) => {
+                    const timeout = setTimeout(() => {
+                        reject(new Error(`Request timed out for block: ${blockNumber}`));
+                    }, this.rpcTimeout ?? 4000);
 
-                let eth: Eth = this.eth;
-                if (retryCount > 0 && this.alternateEth) {
-                    eth = this.alternateEth;
-                }
-
-                (eth.currentProvider as WebsocketProvider).send({
-                    method: "qn_getBlockWithReceipts",
-                    id: Date.now().toString() + blockNumber,
-                    params: [utils.numberToHex(blockNumber)],
-                    jsonrpc: "2.0"
-                }, (error, response) => {
-                    if (error) {
-                        clearTimeout(timeout);
-                        reject(error);
+                    let eth: Eth = this.eth;
+                    if (retryCount > 0 && this.alternateEth) {
+                        await new Promise(r => setTimeout(r, 2000));
+                        eth = this.alternateEth;
                     }
 
-                    clearTimeout(timeout);
-                    resolve(response?.result);
+                    (eth.currentProvider as WebsocketProvider).send({
+                        method: "qn_getBlockWithReceipts",
+                        id: Date.now().toString() + blockNumber,
+                        params: [utils.numberToHex(blockNumber)],
+                        jsonrpc: "2.0"
+                    }, (error, response) => {
+                        if (error) {
+                            clearTimeout(timeout);
+                            reject(error);
+                        }
+
+                        if (!(response?.result)) {
+                            clearTimeout(timeout);
+                            reject(
+                                new Error(
+                                    `null response received for block: ${blockNumber}`
+                                )
+                            );
+                        }
+
+                        clearTimeout(timeout);
+                        resolve(response?.result);
+                    });
                 });
-            });
 
             const transactions: ITransaction[] = [];
 

diff --git a/internal/block_producers/block_producer.ts b/internal/block_producers/block_producer.ts
@@ -89,6 +89,12 @@ export class BlockProducer extends AsynchronousProducer {
         await this.database.connect();
         const metadata = await super.start();
 
+        Logger.info({
+            location: "block_producer",
+            function: "start",
+            message: "Producer started",
+        });
+
         this.on("delivered", async (report: DeliveryReport) => {
             if (report.partition === -1) {
                 const error = new BlockProducerError(

diff --git a/internal/block_subscription/block_subscription.ts b/internal/block_subscription/block_subscription.ts
@@ -71,6 +71,7 @@ export class BlockSubscription extends AbstractBlockSubscription {
                     workerData
                 }
             );
+            worker.setMaxListeners(1000);
 
             worker.on("exit", () => {
                 this.workers[i] = new Worker(
@@ -79,6 +80,7 @@ export class BlockSubscription extends AbstractBlockSubscription {
                         workerData
                     }
                 );
+                this.workers[i].setMaxListeners(1000);
             });
 
             workers.push(worker);

diff --git a/internal/coder/protobuf_coder.ts b/internal/coder/protobuf_coder.ts
@@ -3,6 +3,7 @@ import { CoderError } from "../errors/coder_error.js";
 import protobuf, { Root, Type } from "protobufjs";
 import { ICoder } from "../interfaces/coder.js";
 import { createRequire } from "module";
+import { Logger } from "../logger/logger.js";
 const { load } = protobuf;
 
 /**
@@ -76,6 +77,13 @@ export class Coder implements ICoder {
         }
 
         try {
+            if (this.messageType === "L1StateBlock") {
+                Logger.info({message: "In coder deserialize - for L1StateBlock", data: {
+                    base64: buffer.toString("base64"),
+                    stringData: buffer.toString(),
+                    buffer
+                }});
+            } 
             return this.protobufType.decode(buffer);
         } catch (error) {
             throw new CoderError(