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

@@ -208,6 +208,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);
+    }
   }
 
   /**

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

@@ -349,7 +349,7 @@ export class DictionaryService {
     dataSources: BlockHeightMap<DS[]>,
     buildDictionaryQueryEntries: (dataSources: DS[]) => DictionaryQueryEntry[]
   ): void {
-    this.queriesMap = dataSources.map((dataSources) => buildDictionaryQueryEntries(dataSources));
+    this.queriesMap = dataSources.map(buildDictionaryQueryEntries);
   }
 
   async scopedDictionaryEntries(
@@ -360,10 +360,8 @@ export class DictionaryService {
     const queryDetails = this.queriesMap?.getDetails(startBlockHeight);
     const queryEntry: DictionaryQueryEntry[] = queryDetails?.value ?? [];
 
-    // Update end block if query changes
-    if (queryDetails?.endHeight && queryDetails?.endHeight < queryEndBlock) {
-      queryEndBlock = queryDetails?.endHeight;
-    }
+    queryEndBlock =
+      queryDetails?.endHeight && queryDetails?.endHeight < queryEndBlock ? queryDetails.endHeight : queryEndBlock;
 
     const dict = await this.getDictionary(startBlockHeight, queryEndBlock, scaledBatchSize, queryEntry);
 

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

@@ -243,7 +243,7 @@ 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);

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.spec.ts

@@ -0,0 +1,167 @@
+// Copyright 2020-2023 SubQuery Pte Ltd authors & contributors
+// SPDX-License-Identifier: GPL-3.0
+
+import {NodeConfig} from '../configure';
+import {BaseDsProcessorService} from './ds-processor.service';
+import {DynamicDsService} from './dynamic-ds.service';
+import {BaseProjectService} from './project.service';
+import {ISubqueryProject} from './types';
+
+class TestProjectService extends BaseProjectService<any, any> {
+  packageVersion = '1.0.0';
+
+  async getBlockTimestamp(height: number): Promise<Date> {
+    return Promise.resolve(new Date());
+  }
+
+  onProjectChange(project: any): void {
+    return;
+  }
+}
+
+describe('BaseProjectService', () => {
+  let service: TestProjectService;
+
+  beforeEach(() => {
+    service = new TestProjectService(
+      null as unknown as BaseDsProcessorService,
+      null as unknown as any,
+      null as unknown as any,
+      null as unknown as any,
+      null as unknown as any,
+      {dataSources: []} as unknown as ISubqueryProject<any>,
+      null as unknown as any,
+      null as unknown as any,
+      {unsafe: false} as unknown as NodeConfig,
+      {getDynamicDatasources: jest.fn()} as unknown as DynamicDsService<any>,
+      null as unknown as any,
+      null as unknown as any
+    );
+  });
+
+  it('hasDataSourcesAfterHeight', async () => {
+    service.getDataSources = jest.fn().mockResolvedValue([{endBlock: 100}, {endBlock: 200}, {endBlock: 300}]);
+
+    const result = await service.hasDataSourcesAfterHeight(150);
+    expect(result).toBe(true);
+  });
+
+  it('hasDataSourcesAfterHeight - undefined endBlock', async () => {
+    service.getDataSources = jest.fn().mockResolvedValue([{endBlock: 100}, {endBlock: undefined}]);
+
+    const result = await service.hasDataSourcesAfterHeight(150);
+    expect(result).toBe(true);
+  });
+
+  it('getDataSources', async () => {
+    (service as any).project.dataSources = [
+      {startBlock: 100, endBlock: 200},
+      {startBlock: 1, endBlock: 100},
+    ];
+    (service as any).dynamicDsService.getDynamicDatasources = jest
+      .fn()
+      .mockResolvedValue([{startBlock: 150, endBlock: 250}]);
+
+    const result = await service.getDataSources(175);
+    expect(result).toEqual([
+      {startBlock: 100, endBlock: 200},
+      {startBlock: 150, endBlock: 250},
+    ]);
+  });
+
+  describe('getDatasourceMap', () => {
+    it('should add endBlock heights correctly', () => {
+      (service as any).dynamicDsService.dynamicDatasources = [];
+      (service as any).projectUpgradeService = {
+        projects: [
+          [
+            1,
+            {
+              dataSources: [
+                {startBlock: 1, endBlock: 300},
+                {startBlock: 10, endBlock: 20},
+                {startBlock: 1, endBlock: 100},
+                {startBlock: 50, endBlock: 200},
+                {startBlock: 500},
+              ],
+            },
+          ],
+        ],
+      } as any;
+
+      const result = service.getDataSourcesMap();
+      expect(result.getAll()).toEqual(
+        new Map([
+          [
+            1,
+            [
+              {startBlock: 1, endBlock: 300},
+              {startBlock: 1, endBlock: 100},
+            ],
+          ],
+          [
+            10,
+            [
+              {startBlock: 1, endBlock: 300},
+              {startBlock: 1, endBlock: 100},
+              {startBlock: 10, endBlock: 20},
+            ],
+          ],
+          [
+            21,
+            [
+              {startBlock: 1, endBlock: 300},
+              {startBlock: 1, endBlock: 100},
+            ],
+          ],
+          [
+            50,
+            [
+              {startBlock: 1, endBlock: 300},
+              {startBlock: 1, endBlock: 100},
+              {startBlock: 50, endBlock: 200},
+            ],
+          ],
+          [
+            101,
+            [
+              {startBlock: 1, endBlock: 300},
+              {startBlock: 50, endBlock: 200},
+            ],
+          ],
+          [201, [{startBlock: 1, endBlock: 300}]],
+          [301, []],
+          [500, [{startBlock: 500}]],
+        ])
+      );
+    });
+
+    it('should contain datasources from current project only', () => {
+      (service as any).dynamicDsService.dynamicDatasources = [];
+      (service as any).projectUpgradeService = {
+        projects: [
+          [
+            1,
+            {
+              dataSources: [{startBlock: 1}, {startBlock: 200}],
+            },
+          ],
+          [
+            100,
+            {
+              dataSources: [{startBlock: 100}],
+            },
+          ],
+        ],
+      } as any;
+
+      const result = service.getDataSourcesMap();
+      expect(result.getAll()).toEqual(
+        new Map([
+          [1, [{startBlock: 1}]],
+          [100, [{startBlock: 100}]],
+        ])
+      );
+    });
+  });
+});