improve metrics and persist cursors

Author: saihaj

pnpm-lock.yaml

@@ -6,9 +6,10 @@
     "dev": "tsx --watch ./src/index.mts | pino-pretty"
   },
   "dependencies": {
-    "@substreams/core": "^0.16.0",
-    "@substreams/manifest": "^0.15.0",
-    "@substreams/node": "^0.6.2",
+    "@connectrpc/connect-node": "1.4.0",
+    "@substreams/core": "0.16.0",
+    "@substreams/manifest": "0.15.0",
+    "@substreams/node": "0.6.2",
     "bullmq": "^5.8.1",
     "fets": "^0.8.0",
     "ioredis": "^5.4.1",

substream-listener/package.json

@@ -1,7 +1,6 @@
 import { createRouter, Response } from "fets";
 import { App } from "uWebSockets.js";
 import { isAddress } from "viem";
-import { sendWebhook } from "./send-webhook.mjs";
 import {
   invalidHttpRequests,
   registry,

substream-listener/src/index.mts

@@ -12,6 +12,7 @@ const DEFAULT_LABEL_NAMES = [
   "module_hash",
   "contract_address",
   "output_module",
+  "app_id",
 ] as const;
 
 function calculateHeadBlockTimeDrift(clock: Clock) {
@@ -20,7 +21,6 @@ function calculateHeadBlockTimeDrift(clock: Clock) {
 }
 
 // Counters
-
 export const invalidHttpRequests = new promClient.Counter({
   name: "http_invalid_request",
   help: "The number of invalid HTTP requests received",
@@ -121,6 +121,7 @@ export function onPrometheusMetrics(
     substreamsEndpoint: string;
     contractAddress: string;
     moduleHash: string;
+    appId: string;
   },
 ) {
   manifest?.set(
@@ -132,6 +133,7 @@ export function onPrometheusMetrics(
       module_hash: options.moduleHash,
       contract_address: options.contractAddress,
       output_module: emitter.request.outputModule,
+      app_id: options.appId,
     },
     1,
   );
@@ -146,6 +148,7 @@ export function onPrometheusMetrics(
         module_hash: options.moduleHash,
         contract_address: options.contractAddress,
         output_module: emitter.request.outputModule,
+        app_id: options.appId,
       },
       1,
     );
@@ -157,6 +160,7 @@ export function onPrometheusMetrics(
         module_hash: options.moduleHash,
         contract_address: options.contractAddress,
         output_module: emitter.request.outputModule,
+        app_id: options.appId,
       })
       .inc(1),
   );
@@ -167,6 +171,7 @@ export function onPrometheusMetrics(
         module_hash: options.moduleHash,
         contract_address: options.contractAddress,
         output_module: emitter.request.outputModule,
+        app_id: options.appId,
       })
       .inc(1);
 
@@ -175,6 +180,7 @@ export function onPrometheusMetrics(
         module_hash: options.moduleHash,
         contract_address: options.contractAddress,
         output_module: emitter.request.outputModule,
+        app_id: options.appId,
       })
       ?.inc(block.toBinary().byteLength);
 
@@ -183,6 +189,7 @@ export function onPrometheusMetrics(
         module_hash: options.moduleHash,
         contract_address: options.contractAddress,
         output_module: emitter.request.outputModule,
+        app_id: options.appId,
       },
       1,
     );
@@ -193,6 +200,7 @@ export function onPrometheusMetrics(
           module_hash: options.moduleHash,
           contract_address: options.contractAddress,
           output_module: emitter.request.outputModule,
+          app_id: options.appId,
         },
         Number(block.clock.number),
       );
@@ -202,6 +210,7 @@ export function onPrometheusMetrics(
           module_hash: options.moduleHash,
           contract_address: options.contractAddress,
           output_module: emitter.request.outputModule,
+          app_id: options.appId,
         },
         calculateHeadBlockTimeDrift(block.clock),
       );
@@ -211,6 +220,7 @@ export function onPrometheusMetrics(
           module_hash: options.moduleHash,
           contract_address: options.contractAddress,
           output_module: emitter.request.outputModule,
+          app_id: options.appId,
         },
         Number(block.clock.timestamp?.seconds),
       );

substream-listener/src/prometheus.mts

@@ -3,15 +3,16 @@ import { Redis } from "ioredis";
 import pTimeout from "p-timeout";
 import { type Logger } from "pino";
 import * as prometheus from "./prometheus.mjs";
+import { createGrpcWebTransport } from "@connectrpc/connect-node";
 import {
   applyParams,
+  createAuthInterceptor,
   createModuleHashHex,
   createRegistry,
   createRequest,
 } from "@substreams/core";
 import { readPackage } from "@substreams/manifest";
 import { BlockEmitter } from "@substreams/node";
-import { createNodeTransport } from "@substreams/node/createNodeTransport";
 import { fileURLToPath } from "node:url";
 import path from "node:path";
 import { Svix } from "svix";
@@ -93,7 +94,13 @@ export function createScheduler(config: {
     const worker = new Worker<Input>(
       config.queueName,
       async (job) => {
-        logger.info(
+        const jobLogger = logger.child({
+          jobId: job.id,
+          jobName: job.name,
+        });
+        const cursorKey = `cursor:${job.name}:${job.id}`.toLowerCase();
+
+        jobLogger.info(
           {
             payload: job.data,
           },
@@ -114,7 +121,7 @@ export function createScheduler(config: {
           throw new Error("No modules found in substream package");
         }
 
-        logger.debug(
+        jobLogger.debug(
           { contractAddress },
           "Applying params to substream package",
         );
@@ -127,18 +134,26 @@ export function createScheduler(config: {
           substreamPackage.modules,
           outputModule,
         );
-        logger.debug({ moduleHash }, "Module hash");
+        jobLogger.debug({ moduleHash }, "Module hash");
 
         const registry = createRegistry(substreamPackage);
-        const transport = createNodeTransport(
-          substreamsEndpoint,
-          token,
-          registry,
-        );
+        const transport = createGrpcWebTransport({
+          baseUrl: substreamsEndpoint,
+          httpVersion: "2",
+          interceptors: [createAuthInterceptor(token)],
+          jsonOptions: {
+            typeRegistry: registry,
+          },
+        });
+
+        const startCursor =
+          (await redisConnection?.get(cursorKey)) || undefined;
+        logger.debug({ startCursor }, "Starting from cursor");
         const request = createRequest({
           substreamPackage,
           outputModule,
           startBlockNum: startBlock,
+          startCursor,
         });
 
         // NodeJS Events
@@ -150,16 +165,19 @@ export function createScheduler(config: {
           substreamsEndpoint,
           contractAddress,
           moduleHash,
+          appId,
+        });
+
+        emitter.on("cursor", async (cursor) => {
+          if (cursor) {
+            await redisConnection?.set(cursorKey, cursor);
+          }
         });
 
         // Stream Blocks
         emitter.on("anyMessage", async (message, cursor, clock) => {
           const transfers = (message.transfers || []) as any[];
 
-          logger.debug(
-            { transfers: transfers.length },
-            "Sending transfers to Svix",
-          );
           const events = transfers.map((transfer) => {
             return svix.message.create(appId, {
               eventType: "erc721.transfer",
@@ -173,7 +191,7 @@ export function createScheduler(config: {
           try {
             await Promise.all(events);
           } catch (e) {
-            logger.error({ error: e }, "Error sending events to Svix");
+            jobLogger.error({ error: e }, "Error sending events to Svix");
           }
         });
 
@@ -184,15 +202,15 @@ export function createScheduler(config: {
           // End of Stream
           emitter.on("close", (error) => {
             if (error) {
-              logger.error({ error }, "Error closing stream");
+              jobLogger.error({ error }, "Error closing stream");
               reject(error);
             }
             resolve("Stream closed");
           });
 
           // Fatal Error
           emitter.on("fatalError", (error) => {
-            logger.fatal({ error }, "Fatal error in stream");
+            jobLogger.fatal({ error }, "Fatal error in stream");
             reject(error);
           });
         });