Add reindexing feature

packages/node/src/indexer/mmr.service.ts

@@ -250,4 +250,15 @@ export class MmrService implements OnApplicationShutdown {
       nodes,
     };
   }
+
+  async deleteMmrNode(blockHeight: number, blockOffset: number): Promise<void> {
+    this.fileBasedMmr = await this.ensureFileBasedMmr(this.nodeConfig.mmrPath);
+    const leafIndex = blockHeight - blockOffset - 1;
+    if (leafIndex < 0) {
+      throw new Error(
+        `Target block height must greater equal to ${blockOffset + 1} `,
+      );
+    }
+    await this.fileBasedMmr.delete(leafIndex);
+  }
 }

packages/node/src/indexer/project.service.ts

@@ -81,6 +81,10 @@ export class ProjectService {
       }
 
       this._startHeight = await this.getStartHeight();
+
+      if (argv.reindex !== undefined) {
+        await this.reindex(argv.reindex);
+      }
     } else {
       this.metadataRepo = MetadataFactory(this.sequelize, this.schema);
 
@@ -369,4 +373,33 @@ export class ProjectService {
           this.apiService.getApi().runtimeVersion.specName.toString(),
     );
   }
+
+  private async reindex(targetBlockHeight: number): Promise<void> {
+    const lastProcessedHeight = await this.getLastProcessedHeight();
+    if (!this.storeService.historical) {
+      logger.warn('Unable to reindex, historical state not enabled');
+      return;
+    }
+    if (!lastProcessedHeight || lastProcessedHeight < targetBlockHeight) {
+      logger.warn(
+        `Skipping reindexing to block ${targetBlockHeight}: current indexing height ${lastProcessedHeight} is behind requested block`,
+      );
+      return;
+    }
+    logger.info(`Reindexing to block: ${targetBlockHeight}`);
+    const transaction = await this.sequelize.transaction();
+    try {
+      await this.storeService.rewind(argv.reindex, transaction);
+
+      const blockOffset = await this.getMetadataBlockOffset();
+      if (blockOffset) {
+        await this.mmrService.deleteMmrNode(targetBlockHeight + 1, blockOffset);
+      }
+      await transaction.commit();
+    } catch (err) {
+      logger.error(err, 'Reindexing failed');
+      await transaction.rollback();
+      throw err;
+    }
+  }
 }

packages/node/src/indexer/store.service.ts

@@ -84,7 +84,7 @@ export class StoreService {
   private metaDataRepo: MetadataRepo;
   private operationStack: StoreOperations;
   private blockHeight: number;
-  private historical: boolean;
+  historical: boolean;
 
   constructor(private sequelize: Sequelize, private config: NodeConfig) {}
 
@@ -577,6 +577,59 @@ group by
     );
   }
 
+  async rewind(
+    targetBlockHeight: number,
+    transaction: Transaction,
+  ): Promise<void> {
+    for (const model of Object.values(this.sequelize.models)) {
+      if ('__block_range' in model.getAttributes()) {
+        await model.destroy({
+          transaction,
+          hooks: false,
+          where: this.sequelize.where(
+            this.sequelize.fn('lower', this.sequelize.col('_block_range')),
+            Op.gt,
+            targetBlockHeight,
+          ),
+        });
+        await model.update(
+          {
+            __block_range: this.sequelize.fn(
+              'int8range',
+              this.sequelize.fn('lower', this.sequelize.col('_block_range')),
+              null,
+            ),
+          },
+          {
+            transaction,
+            hooks: false,
+            where: {
+              __block_range: {
+                [Op.contains]: targetBlockHeight,
+              },
+            },
+          },
+        );
+      }
+    }
+    await this.setMetadata('lastProcessedHeight', targetBlockHeight, {
+      transaction,
+    });
+    if (this.config.proofOfIndex) {
+      await this.poiRepo.destroy({
+        transaction,
+        where: {
+          id: {
+            [Op.gt]: targetBlockHeight,
+          },
+        },
+      });
+      await this.setMetadata('lastPoiHeight', targetBlockHeight, {
+        transaction,
+      });
+    }
+  }
+
   getStore(): Store {
     return {
       get: async (entity: string, id: string): Promise<Entity | undefined> => {

packages/node/src/yargs.ts

@@ -151,6 +151,11 @@ export function getYargsOption() {
       describe: 'Disable storing historical state entities',
       type: 'boolean',
     },
+    reindex: {
+      demandOption: false,
+      describe: 'Reindex to specified block height',
+      type: 'number',
+    },
     workers: {
       alias: 'w',
       demandOption: false,