Endblock feature

packages/common-substrate/src/project/models.ts

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import {RegisteredTypes, RegistryTypes, OverrideModuleType, OverrideBundleType} from '@polkadot/types/types';
-import {BlockFilterImpl, ProcessorImpl} from '@subql/common';
+import {BaseDataSource, BlockFilterImpl, ProcessorImpl} from '@subql/common';
 import {
   SubstrateBlockFilter,
   SubstrateBlockHandler,
@@ -24,7 +24,6 @@ import {
   IsArray,
   IsBoolean,
   IsEnum,
-  IsInt,
   IsOptional,
   IsString,
   IsObject,
@@ -149,15 +148,12 @@ export class CustomMapping implements BaseMapping<SubstrateCustomHandler> {
   file: string;
 }
 
-export class RuntimeDataSourceBase implements SubstrateRuntimeDatasource {
+export class RuntimeDataSourceBase extends BaseDataSource implements SubstrateRuntimeDatasource {
   @IsEnum(SubstrateDatasourceKind, {groups: [SubstrateDatasourceKind.Runtime]})
   kind: SubstrateDatasourceKind.Runtime;
   @Type(() => RuntimeMapping)
   @ValidateNested()
   mapping: RuntimeMapping;
-  @IsOptional()
-  @IsInt()
-  startBlock?: number;
 }
 
 export class FileReferenceImpl implements FileReference {
@@ -166,16 +162,14 @@ export class FileReferenceImpl implements FileReference {
 }
 
 export class CustomDataSourceBase<K extends string, M extends CustomMapping, O = any>
+  extends BaseDataSource
   implements SubstrateCustomDatasource<K, M, O>
 {
   @IsString()
   kind: K;
   @Type(() => CustomMapping)
   @ValidateNested()
   mapping: M;
-  @IsOptional()
-  @IsInt()
-  startBlock?: number;
   @Type(() => FileReferenceImpl)
   @ValidateNested({each: true})
   assets: Map<string, FileReference>;

packages/common/src/project/utils.ts

@@ -4,7 +4,7 @@
 import fs from 'fs';
 import os from 'os';
 import path from 'path';
-import {FileReference, MultichainProjectManifest, ProjectRootAndManifest} from '@subql/types-core';
+import {BaseDataSource, FileReference, MultichainProjectManifest, ProjectRootAndManifest} from '@subql/types-core';
 import {
   registerDecorator,
   validateSync,
@@ -294,3 +294,17 @@ export class FileReferenceImp<T> implements ValidatorConstraintInterface {
 }
 
 export const tsProjectYamlPath = (tsManifestEntry: string) => tsManifestEntry.replace('.ts', '.yaml');
+
+@ValidatorConstraint({async: false})
+export class IsEndBlockGreater implements ValidatorConstraintInterface {
+  validate(endBlock: number, args: ValidationArguments) {
+    const object = args.object as BaseDataSource;
+    return object.startBlock !== undefined && object.endBlock !== undefined
+      ? object.endBlock >= object.startBlock
+      : true;
+  }
+
+  defaultMessage(args: ValidationArguments) {
+    return 'End block must be greater than or equal to start block';
+  }
+}

packages/common/src/project/versioned/base.ts

@@ -3,9 +3,19 @@
 
 import {FileReference, ParentProject, Processor} from '@subql/types-core';
 import {plainToInstance, Type} from 'class-transformer';
-import {Allow, Equals, IsObject, IsOptional, IsString, ValidateNested, validateSync} from 'class-validator';
+import {
+  Allow,
+  Equals,
+  IsInt,
+  IsObject,
+  IsOptional,
+  IsString,
+  Validate,
+  ValidateNested,
+  validateSync,
+} from 'class-validator';
 import yaml from 'js-yaml';
-import {toJsonObject} from '../utils';
+import {IsEndBlockGreater, toJsonObject} from '../utils';
 import {ParentProjectModel} from './v1_0_0/models';
 
 export abstract class ProjectManifestBaseImpl<D extends BaseDeploymentV1_0_0> {
@@ -78,3 +88,13 @@ export class BaseDeploymentV1_0_0 {
     });
   }
 }
+
+export class BaseDataSource {
+  @IsOptional()
+  @IsInt()
+  startBlock?: number;
+  @Validate(IsEndBlockGreater)
+  @IsOptional()
+  @IsInt()
+  endBlock?: number;
+}

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

@@ -207,6 +207,12 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS> implements IBloc
       // Flush all data from cache and wait
       await this.storeCacheService.flushCache(false, true);
     }
+
+    if (!(await this.projectService.hasDataSourcesAfterHeight(height))) {
+      logger.info(`All data sources have been processed up to block number ${height}. Exiting gracefully...`);
+      await this.storeCacheService.flushCache(false, true);
+      process.exit(0);
+    }
   }
 
   // First creation of POI

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

@@ -243,10 +243,15 @@ export abstract class BaseFetchService<
         startBlockHeight >= this.dictionaryService.startHeight &&
         startBlockHeight < this.latestFinalizedHeight
       ) {
-        /* queryEndBlock needs to be limited by the latest height.
+        /* queryEndBlock needs to be limited by the latest height or the maximum value of endBlock in datasources.
          * Dictionaries could be in the future depending on if they index unfinalized blocks or the node is using an RPC endpoint that is behind.
          */
-        const queryEndBlock = Math.min(startBlockHeight + DICTIONARY_MAX_QUERY_SIZE, this.latestFinalizedHeight);
+        const maxEndBlock = await this.projectService.maxEndBlockHeight();
+        const queryEndBlock = Math.min(
+          startBlockHeight + DICTIONARY_MAX_QUERY_SIZE,
+          this.latestFinalizedHeight,
+          maxEndBlock
+        );
         try {
           const dictionary = await this.dictionaryService.scopedDictionaryEntries(
             startBlockHeight,

packages/node-core/src/indexer/indexer.manager.ts

@@ -146,8 +146,13 @@ export abstract class BaseIndexerManager<
   private assertDataSources(ds: DS[], blockHeight: number) {
     if (!ds.length) {
       logger.error(
-        `Your start block of all the datasources is greater than the current indexed block height in your database. Either change your startBlock (project.yaml) to <= ${blockHeight}
-         or delete your database and start again from the currently specified startBlock`
+        `Issue detected with data sources: \n
+        Either all data sources have a 'startBlock' greater than the current indexed block height (${blockHeight}), 
+        or they have an 'endBlock' less than the current block. \n
+        Solution options: \n
+        1. Adjust 'startBlock' in project.yaml to be less than or equal to ${blockHeight}, 
+           and 'endBlock' to be greater than or equal to ${blockHeight}. \n
+        2. Delete your database and start again with the currently specified 'startBlock' and 'endBlock'.`
       );
       process.exit(1);
     }

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

@@ -256,13 +256,28 @@ export abstract class BaseProjectService<API extends IApi, DS extends BaseDataSo
     return [...dataSources, ...dynamicDs];
   }
 
+  async hasDataSourcesAfterHeight(height: number): Promise<boolean> {
+    const dataSources = await this.getDataSources();
+    return dataSources.some((ds) => ds.endBlock !== undefined && ds.endBlock > height);
+  }
+
+  async maxEndBlockHeight(): Promise<number> {
+    const dataSources = await this.getDataSources();
+    const endBlocks = dataSources.map((ds) => (ds.endBlock !== undefined ? ds.endBlock : Number.MAX_SAFE_INTEGER));
+    return Math.max(...endBlocks);
+  }
+
   // This gets used when indexing blocks, it needs to be async to ensure dynamicDs is updated within workers
   async getDataSources(blockHeight?: number): Promise<DS[]> {
     const dataSources = this.project.dataSources;
     const dynamicDs = await this.dynamicDsService.getDynamicDatasources();
 
     return [...dataSources, ...dynamicDs].filter(
-      (ds) => blockHeight === undefined || (ds.startBlock !== undefined && ds.startBlock <= blockHeight)
+      (ds) =>
+        blockHeight === undefined ||
+        (ds.startBlock !== undefined &&
+          ds.startBlock <= blockHeight &&
+          (ds.endBlock === undefined || ds.endBlock >= blockHeight))
     );
   }
 

packages/node-core/src/indexer/types.ts

@@ -48,4 +48,6 @@ export interface IProjectService<DS> {
   getDataSources(blockHeight?: number): Promise<DS[]>;
   getStartBlockFromDataSources(): number;
   getDataSourcesMap(): BlockHeightMap<DS[]>;
+  hasDataSourcesAfterHeight(height: number): Promise<boolean>;
+  maxEndBlockHeight(): Promise<number>;
 }

packages/types-core/src/project/versioned/base.ts

@@ -15,6 +15,11 @@ export interface BaseDataSource<H extends BaseHandler = BaseHandler, M extends B
    * @default 1
    */
   startBlock?: number;
+  /**
+   * The ending block number for the datasource (optional).
+   * @type {number}
+   */
+  endBlock?: number;
   /**
    * The mapping associated with the datasource.
    * This contains the handlers.
@@ -69,4 +74,5 @@ export interface TemplateBase {
   name: string;
 }
 
-export type BaseTemplateDataSource<DS extends BaseDataSource = BaseDataSource> = Omit<DS, 'startBlock'> & TemplateBase;
+export type BaseTemplateDataSource<DS extends BaseDataSource = BaseDataSource> = Omit<DS, 'startBlock' | 'endBlock'> &
+  TemplateBase;