Merge pull request #204 from utxostack/fix/cache-n-dataloader

refactor(indexer): improve asset indexing and health check

backend/src/constants.ts

@@ -19,6 +19,7 @@ export const BtcTestnetTypeMap: Record<NetworkType.testnet | NetworkType.signet,
 };
 
 export const CKB_MIN_SAFE_CONFIRMATIONS = 24;
+export const CKB_ONE_DAY_BLOCKS = 8640;
 export const CKB_CHAIN_ID = 1;
 
 export const XUDT_TYPESCRIPTS = {

backend/src/core/indexer/flow/assets.flow.ts

@@ -1,7 +1,7 @@
 import { Logger } from '@nestjs/common';
 import { AssetType, Chain } from '@prisma/client';
 import { EventEmitter } from 'node:events';
-import { CKB_MIN_SAFE_CONFIRMATIONS } from 'src/constants';
+import { CKB_MIN_SAFE_CONFIRMATIONS, CKB_ONE_DAY_BLOCKS } from 'src/constants';
 import { BlockchainService } from 'src/core/blockchain/blockchain.service';
 import { PrismaService } from 'src/core/database/prisma/prisma.service';
 import { IndexerQueueService } from '../indexer.queue';
@@ -24,6 +24,21 @@ export class IndexerAssetsFlow extends EventEmitter {
   }
 
   public async start() {
+    const latestAsset = await this.getLatestAsset();
+    if (latestAsset) {
+      this.logger.log(`Latest asset block number: ${latestAsset.blockNumber}`);
+      const tipBlockNumber = await this.blockchainService.getTipBlockNumber();
+      if (
+        tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS - latestAsset.blockNumber <
+        CKB_ONE_DAY_BLOCKS
+      ) {
+        this.logger.log(`Latest asset is near tip block number, skip indexing assets...`);
+        this.startBlockAssetsIndexing();
+        this.setupBlockAssetsIndexedListener();
+        return;
+      }
+    }
+
     const assetTypeScripts = await this.prismaService.assetType.findMany({
       where: { chainId: this.chain.id },
     });
@@ -32,6 +47,15 @@ export class IndexerAssetsFlow extends EventEmitter {
     this.setupAssetIndexedListener(assetTypeScripts.length);
   }
 
+  private async getLatestAsset() {
+    const latestAsset = await this.prismaService.asset.findFirst({
+      select: { blockNumber: true },
+      where: { chainId: this.chain.id },
+      orderBy: { blockNumber: 'desc' },
+    });
+    return latestAsset;
+  }
+
   private async indexAssets(assetType: AssetType) {
     const cursor = await this.indexerQueueService.getLatestAssetJobCursor(assetType);
     if (cursor === '0x') {
@@ -62,7 +86,7 @@ export class IndexerAssetsFlow extends EventEmitter {
   private async startBlockAssetsIndexing() {
     const tipBlockNumber = await this.blockchainService.getTipBlockNumber();
 
-    let latestIndexedBlockNumber = await this.indexerQueueService.getLatestIndexedBlock(
+    let latestIndexedBlockNumber = await this.indexerQueueService.getLatestIndexedAssetsBlock(
       this.chain.id,
     );
     if (!latestIndexedBlockNumber) {
@@ -75,7 +99,7 @@ export class IndexerAssetsFlow extends EventEmitter {
     }
     const targetBlockNumber = tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS;
     if (targetBlockNumber <= latestIndexedBlockNumber) {
-      this.emit(IndexerAssetsEvent.BlockAssetsIndexed);
+      this.emit(IndexerAssetsEvent.BlockAssetsIndexed, latestIndexedBlockNumber);
       return;
     }
 

backend/src/core/indexer/indexer.health.ts

@@ -60,7 +60,16 @@ export class IndexerHealthIndicator extends HealthIndicator {
     const blockchainService = this.blockchainServiceFactory.getService(CKB_CHAIN_ID);
     const tipBlockNumber = await blockchainService.getTipBlockNumber();
     const targetBlockNumber = tipBlockNumber - CKB_MIN_SAFE_CONFIRMATIONS;
-    const currentBlockNumber = await this.indexerQueueService.getLatestIndexedBlock(CKB_CHAIN_ID);
+
+    let currentBlockNumber = await this.indexerQueueService.getLatestIndexedAssetsBlock(CKB_CHAIN_ID);
+    if (!currentBlockNumber) {
+      const latestAsset = await this.prismaService.asset.findFirst({
+        select: { blockNumber: true },
+        where: { chainId: CKB_CHAIN_ID },
+        orderBy: { blockNumber: 'desc' },
+      });
+      currentBlockNumber = latestAsset?.blockNumber;
+    }
 
     const isHealthy =
       !!currentBlockNumber && currentBlockNumber >= targetBlockNumber - INDEEXR_HEALTH_THRESHOLD;

backend/src/core/indexer/indexer.queue.ts

@@ -76,13 +76,17 @@ export class IndexerQueueService {
     await this.cacheManager.set(`${INDEXER_ASSETS_QUEUE}:${typeHash}`, cursor || '');
   }
 
-  public async getLatestIndexedBlock(chainId: number) {
+  public async getLatestIndexedAssetsBlock(chainId: number) {
     const blockNumber = await this.cacheManager.get<number>(
       `${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`,
     );
     return blockNumber;
   }
 
+  public async setLatestIndexedAssetsBlock(chainId: number, blockNumber: number) {
+    await this.cacheManager.set(`${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`, blockNumber);
+  }
+
   public async addBlockAssetsJob(data: IndexerBlockAssetsJobData) {
     const { chainId, blockNumber } = data;
     const params = new URLSearchParams();
@@ -95,7 +99,7 @@ export class IndexerQueueService {
       `Added block assets job ${jobId} for chain ${chainId} with block number ${blockNumber}`,
     );
     await this.blockAssetsQueue.add(jobId, data, { jobId });
-    await this.cacheManager.set(`${INDEXER_BLOCK_ASSETS_QUEUE}:${chainId}`, blockNumber);
+    await this.setLatestIndexedAssetsBlock(chainId, blockNumber);
   }
 
   public async addLockJob(data: IndexerLockJobData) {

backend/src/core/indexer/processor/block-assets.processor.ts

@@ -107,7 +107,7 @@ export class IndexerBlockAssetsProcessor extends WorkerHost {
     } else {
       const indexerServiceFactory = this.moduleRef.get(IndexerServiceFactory);
       const indexerService = await indexerServiceFactory.getService(chainId);
-      indexerService.assetsFlow.emit(IndexerAssetsEvent.BlockAssetsIndexed, block);
+      indexerService.assetsFlow.emit(IndexerAssetsEvent.BlockAssetsIndexed, blockNumber);
       return;
     }
   }

backend/src/filters/all-exceptions.filter.ts

@@ -11,12 +11,15 @@ export class AllExceptionsFilter extends SentryGlobalGraphQLFilter {
   }
 
   catch(exception: unknown, host: ArgumentsHost) {
-    const ctx = host.switchToHttp();
-    const request = ctx.getRequest();
-    if (SKIP_REQUEST_URLS.includes(request.url)) {
-      const response = (exception as HttpException).getResponse();
-      this.httpAdapterHost.httpAdapter.reply(ctx.getResponse(), response, 200);
-      return;
+    const type = host.getType();
+    if (type === 'http') {
+      const ctx = host.switchToHttp();
+      const request = ctx.getRequest();
+      if (SKIP_REQUEST_URLS.includes(request.url)) {
+        const response = (exception as HttpException).getResponse();
+        this.httpAdapterHost.httpAdapter.reply(ctx.getResponse(), response, 200);
+        return;
+      }
     }
 
     super.catch(exception, host);