Rework poi (#1966)

* migration

poi creation

add indexes

tidy up

fix index query

tidy up

logger, workers and reindex

clean mmr

clean mmr

test and add validation

remove logs

drop mmr table

fix up

fix up test

Update packages/common/src/constants.ts

Co-authored-by: Scott Twiname <[email protected]>

Update packages/node-core/src/indexer/poi/poi.service.ts

Co-authored-by: Scott Twiname <[email protected]>

Update packages/node-core/src/indexer/poi/poi.service.ts

Co-authored-by: Scott Twiname <[email protected]>

Update packages/node-core/src/indexer/poi/poi.service.ts

Co-authored-by: Scott Twiname <[email protected]>

Update packages/node-core/src/indexer/poi/poi.service.ts

Co-authored-by: Scott Twiname <[email protected]>

catch error and exit

fix

add more test

* fix

* lint

* fix test

* update

* comment

* use injection

* private

Author: jiqiang90

packages/common/src/constants.ts

@@ -14,10 +14,8 @@ export const IPFS_REGEX = /^ipfs:\/\//i;
 // MANIFEST
 export const RUNNER_REGEX = /(\^?)(\d|x|\*)+\.(\d|x|\*)+\.(\d|x|\*)+/;
 
-//MMR
-export const DEFAULT_WORD_SIZE = 32;
-export const DEFAULT_LEAF = Buffer.from('0000000000000000000000000000000000000000000000000000000000000001', 'hex');
-export const MMR_AWAIT_TIME = 2;
+// POI
+export const POI_AWAIT_TIME = 2; // seconds
 
 // NETWORK
 export enum NETWORK_FAMILY {

packages/node-core/package.json

@@ -27,7 +27,6 @@
     "@subql/testing": "workspace:*",
     "@subql/types": "workspace:*",
     "@subql/utils": "workspace:*",
-    "@subql/x-merkle-mountain-range": "^2.0.0-0.1.3",
     "@subql/x-sequelize": "6.32.0-0.0.2",
     "@willsoto/nestjs-prometheus": "^5.1.1",
     "async-lock": "^1.4.0",

packages/node-core/src/configure/NodeConfig.ts

@@ -12,11 +12,6 @@ import {assign} from '../utils/object';
 
 const logger = getLogger('configure');
 
-export enum MmrStoreType {
-  File = 'file',
-  Postgres = 'postgres',
-}
-
 export interface IConfig {
   readonly subquery: string;
   readonly subqueryName?: string;
@@ -36,8 +31,6 @@ export interface IConfig {
   readonly indexCountLimit: number;
   readonly timestampField: boolean;
   readonly proofOfIndex: boolean;
-  readonly mmrStoreType: MmrStoreType;
-  readonly mmrPath?: string;
   readonly ipfs?: string;
   readonly dictionaryTimeout: number;
   readonly workers?: number;
@@ -75,7 +68,6 @@ const DEFAULT_CONFIG = {
   indexCountLimit: 10,
   timestampField: true,
   proofOfIndex: false,
-  mmrStoreType: MmrStoreType.Postgres,
   dictionaryTimeout: 30,
   profiler: false,
   subscription: false,
@@ -219,17 +211,10 @@ export class NodeConfig implements IConfig {
     return this._config.proofOfIndex;
   }
 
-  get mmrStoreType(): MmrStoreType {
-    return this._config.mmrStoreType;
-  }
-
   get dictionaryTimeout(): number {
     return this._config.dictionaryTimeout;
   }
 
-  get mmrPath(): string {
-    return this._config.mmrPath ?? `.mmr/${this.subqueryName}.mmr`;
-  }
   get ipfs(): string | undefined {
     return this._config.ipfs;
   }

packages/node-core/src/events.ts

@@ -18,7 +18,7 @@ export enum IndexerEvent {
 }
 
 export enum PoiEvent {
-  LastPoiWithMmr = 'last_poi_with_mmr',
+  LatestSyncedPoi = 'poi_synced',
   PoiTarget = 'poi_target',
 }
 
@@ -48,21 +48,3 @@ export interface NetworkMetadataPayload {
   specName: string;
   genesisHash: string;
 }
-
-export interface MmrPayload {
-  offset: number;
-  height: number;
-  hash: string; //the node hash
-  mmrRoot: string;
-}
-
-export interface MmrProof {
-  digest: string;
-  leafLength: number;
-  nodes: MmrNode[];
-}
-
-export interface MmrNode {
-  node: string;
-  hash: string;
-}

packages/node-core/src/indexer/benchmark.service.ts

@@ -104,8 +104,8 @@ export class PoiBenchmarkService extends BaseBenchmarkService {
     }
   }
 
-  @OnEvent(PoiEvent.LastPoiWithMmr)
-  handleLastPoiWithMmr(blockPayload: ProcessBlockPayload): void {
+  @OnEvent(PoiEvent.LatestSyncedPoi)
+  handleLatestSyncedPoi(blockPayload: ProcessBlockPayload): void {
     this.currentProcessingHeight = blockPayload.height;
     this.currentProcessingTimestamp = blockPayload.timestamp;
   }

packages/node-core/src/indexer/blockDispatcher/base-block-dispatcher.ts

@@ -167,22 +167,15 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBloc
     const operationHash = this.storeService.getOperationMerkleRoot();
     const {blockHash, dynamicDsCreated, reindexBlockHeight} = processBlockResponse;
 
-    await this.updatePOI(height, blockHash, operationHash);
+    this.createPOI(height, blockHash, operationHash);
 
     if (reindexBlockHeight !== null && reindexBlockHeight !== undefined) {
       await this.rewind(reindexBlockHeight);
       this.setLatestProcessedHeight(reindexBlockHeight);
     } else {
       this.updateStoreMetadata(height);
       if (this.nodeConfig.proofOfIndex && !isNullMerkelRoot(operationHash)) {
-        // We only check if it is undefined, need to be caution here when blockOffset is 0
-        if (this.projectService.blockOffset === undefined) {
-          // Which means during project init, it has not found offset and set value
-          this.storeCacheService.metadata.set('blockOffset', height - 1);
-        }
-        // this will return if project service blockOffset already exist
-        // dont await this, this starts a loop
-        void this.projectService.setBlockOffset(height - 1);
+        await this.poiService.ensureGenesisPoi(height);
       }
       if (dynamicDsCreated) {
         await this.onDynamicDsCreated(height);
@@ -208,27 +201,20 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBloc
     }
   }
 
-  private async updatePOI(height: number, blockHash: string, operationHash: Uint8Array): Promise<void> {
+  // First creation of POI
+  private createPOI(height: number, blockHash: string, operationHash: Uint8Array): void {
     if (!this.nodeConfig.proofOfIndex) {
       return;
     }
-    //check if operation is null, then poi will not be inserted
     if (!u8aEq(operationHash, NULL_MERKEL_ROOT)) {
-      const poiBlock = PoiBlock.create(
-        height,
-        blockHash,
-        operationHash,
-        await this.poiService.getLatestPoiBlockHash(),
-        this.project.id
-      );
+      const poiBlock = PoiBlock.create(height, blockHash, operationHash, this.project.id);
       // This is the first creation of POI
       this.poi.bulkUpsert([poiBlock]);
+      this.storeCacheService.metadata.setBulk([{key: 'lastCreatedPoiHeight', value: height}]);
       this.eventEmitter.emit(PoiEvent.PoiTarget, {
         height,
         timestamp: Date.now(),
       });
-      this.poiService.setLatestPoiBlockHash(poiBlock.hash);
-      this.storeCacheService.metadata.set('lastPoiHeight', height);
     }
   }
 
@@ -248,7 +234,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBloc
   private get poi(): CachePoiModel {
     const poi = this.storeCacheService.poi;
     if (!poi) {
-      throw new Error('MMR service expected POI but it was not found');
+      throw new Error('Poi service expected poi repo but it was not found');
     }
     return poi;
   }

packages/node-core/src/indexer/entities/Metadata.entity.ts

@@ -19,15 +19,17 @@ export interface MetadataKeys {
   runnerQuery: string;
   runnerQueryVersion: string;
   specName: string;
-  lastPoiHeight: number;
   lastFinalizedVerifiedHeight: number;
   indexerHealthy: boolean;
   targetHeight: number;
   dynamicDatasources: string;
   unfinalizedBlocks: string;
   schemaMigrationCount: number;
   deployments: string;
-  latestPoiWithMmr: string;
+  lastCreatedPoiHeight: number;
+  latestSyncedPoiHeight: number;
+  latestPoiWithMmr: string; // Deprecated, keep for poi migration
+  lastPoiHeight: string; // Deprecated, keep for poi migration
 }
 
 export interface Metadata {

packages/node-core/src/indexer/entities/Mmr.entitiy.ts

@@ -1,18 +1,22 @@
 // Copyright 2020-2023 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
-import {BuildOptions, DataTypes, Model, Sequelize} from '@subql/x-sequelize';
+import {BuildOptions, DataTypes, Model, Op, Sequelize} from '@subql/x-sequelize';
 
 export interface ProofOfIndex {
   id: number; //blockHeight
-  chainBlockHash: Uint8Array;
-  hash: Uint8Array;
-  parentHash?: Uint8Array;
-  operationHashRoot: Uint8Array;
-  mmrRoot?: Uint8Array;
+  chainBlockHash: Uint8Array | null;
+  hash: Uint8Array | undefined;
+  parentHash: Uint8Array | undefined;
+  operationHashRoot: Uint8Array | null;
   projectId?: string;
 }
 
+export interface SyncedProofOfIndex extends ProofOfIndex {
+  hash: Uint8Array;
+  parentHash: Uint8Array;
+}
+
 export interface PoiModel extends Model<ProofOfIndex>, ProofOfIndex {}
 
 export type PoiRepo = typeof Model & {
@@ -46,11 +50,6 @@ export function PoiFactoryDeprecate(sequelize: Sequelize, schema: string): PoiRe
         type: DataTypes.BLOB,
         allowNull: false,
       },
-      mmrRoot: {
-        type: DataTypes.BLOB,
-        allowNull: true,
-        unique: true,
-      },
       projectId: {
         type: DataTypes.STRING,
         allowNull: false,
@@ -75,33 +74,27 @@ export function PoiFactory(sequelize: Sequelize, schema: string): PoiRepo {
       },
       chainBlockHash: {
         type: DataTypes.BLOB,
-        allowNull: false,
+        allowNull: true,
         unique: true,
       },
       hash: {
         type: DataTypes.BLOB,
-        allowNull: false,
+        allowNull: true,
         unique: true,
       },
       parentHash: {
         type: DataTypes.BLOB,
-        allowNull: false,
+        allowNull: true,
         unique: true,
       },
       operationHashRoot: {
-        type: DataTypes.BLOB,
-        allowNull: false,
-      },
-      mmrRoot: {
         type: DataTypes.BLOB,
         allowNull: true,
-        unique: true,
       },
     },
     {
       freezeTableName: true,
       schema: schema,
-      indexes: [{fields: ['hash']}],
     }
   );
 }