feat: implemented dynamic datasource removing themseleves and adding them at end blocks

packages/node-core/src/indexer/dynamic-ds.service.ts

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import {Inject, Injectable} from '@nestjs/common';
-import {BaseCustomDataSource, BaseDataSource, BaseTemplateDataSource} from '@subql/types-core';
+import {BaseCustomDataSource, BaseDataSource, BaseTemplateDataSource, DynamicDatasourceInfo} from '@subql/types-core';
 import {Transaction} from '@subql/x-sequelize';
 import {cloneDeep} from 'lodash';
 import {IBlockchainService} from '../blockchain.service';
@@ -19,12 +19,16 @@ export interface DatasourceParams {
   templateName: string;
   args?: Record<string, unknown>;
   startBlock: number;
+  endBlock?: number;
 }
 
 export interface IDynamicDsService<DS> {
   dynamicDatasources: DS[];
   createDynamicDatasource(params: DatasourceParams): Promise<DS>;
+  destroyDynamicDatasource(templateName: string, currentBlockHeight: number, index: number): Promise<void>;
   getDynamicDatasources(forceReload?: boolean): Promise<DS[]>;
+  getDynamicDatasourcesByTemplate(templateName: string): DynamicDatasourceInfo[];
+  getDatasourceParamByIndex(index: number): DatasourceParams | undefined;
 }
 
 @Injectable()
@@ -91,6 +95,92 @@ export class DynamicDsService<DS extends BaseDataSource = BaseDataSource, P exte
     }
   }
 
+  /**
+   * Get all active (non-destroyed) dynamic datasources for a specific template.
+   *
+   * @param templateName - The name of the template to filter by
+   * @returns Array of datasource info objects with global indices. The `index` field
+   *          represents the global position in the internal datasource array and should
+   *          be used when calling `destroyDynamicDatasource()`.
+   */
+  getDynamicDatasourcesByTemplate(templateName: string): DynamicDatasourceInfo[] {
+    if (!this._datasourceParams) {
+      throw new Error('DynamicDsService has not been initialized');
+    }
+
+    const matchingDatasources = this._datasourceParams
+      .map((params, globalIndex) => ({params, globalIndex}))
+      .filter(({params}) => params.templateName === templateName && params.endBlock === undefined);
+
+    return matchingDatasources.map(({globalIndex, params}) => ({
+      index: globalIndex,
+      templateName: params.templateName,
+      startBlock: params.startBlock,
+      endBlock: params.endBlock,
+      args: params.args,
+    }));
+  }
+
+  /**
+   * Get datasource parameters by global index.
+   *
+   * @param index - Global index in the internal datasource parameters array
+   * @returns DatasourceParams if found, undefined otherwise
+   */
+  getDatasourceParamByIndex(index: number): DatasourceParams | undefined {
+    return this._datasourceParams?.[index];
+  }
+
+  async destroyDynamicDatasource(
+    templateName: string,
+    currentBlockHeight: number,
+    index: number,
+    tx?: Transaction
+  ): Promise<void> {
+    if (!this._datasources || !this._datasourceParams) {
+      throw new Error('DynamicDsService has not been initialized');
+    }
+
+    // Get the datasource at the global index
+    const dsParam = this._datasourceParams[index];
+
+    // Validate datasource exists
+    if (!dsParam) {
+      throw new Error(
+        `Index ${index} is out of bounds. There are ${this._datasourceParams.length} datasource(s) in total`
+      );
+    }
+
+    // Validate it matches the template name and is not already destroyed
+    if (dsParam.templateName !== templateName) {
+      throw new Error(
+        `Datasource at index ${index} has template name "${dsParam.templateName}", not "${templateName}"`
+      );
+    }
+
+    if (dsParam.endBlock !== undefined) {
+      throw new Error(`Dynamic datasource at index ${index} is already destroyed`);
+    }
+
+    // Update the datasource params
+    const updatedParams = {...dsParam, endBlock: currentBlockHeight};
+    this._datasourceParams[index] = updatedParams;
+
+    // Update the datasource object if it exists
+    // Note: _datasources and _datasourceParams arrays should always be in sync.
+    // If the index is valid for params, it must also be valid for datasources.
+    const datasource = this._datasources[index];
+    if (!datasource) {
+      throw new Error(`Datasources array out of sync with params at index ${index}`);
+    }
+    // Set endBlock on the datasource object
+    datasource.endBlock = currentBlockHeight;
+
+    await this.metadata.set(METADATA_KEY, this._datasourceParams, tx);
+
+    logger.info(`Destroyed dynamic datasource "${templateName}" at block ${currentBlockHeight}`);
+  }
+
   // Not force only seems to be used for project changes
   async getDynamicDatasources(forceReload?: boolean): Promise<DS[]> {
     // Workers should not cache this result in order to keep in sync
@@ -117,19 +207,19 @@ export class DynamicDsService<DS extends BaseDataSource = BaseDataSource, P exte
    *
    * This will throw if the template cannot be found by name.
    *
-   * Inserts the startBlock into the template.
+   * Inserts the startBlock and optionally endBlock into the template.
    * */
-  protected getTemplate(templateName: string, startBlock?: number): DS {
+  protected getTemplate(templateName: string, startBlock?: number, endBlock?: number): DS {
     const t = (this.project.templates ?? []).find((t) => t.name === templateName);
     if (!t) {
       throw new Error(`Unable to find matching template in project for name: "${templateName}"`);
     }
     const {name, ...template} = cloneDeep(t);
-    return {...template, startBlock} as DS;
+    return {...template, startBlock, endBlock} as DS;
   }
 
   private async getDatasource(params: DatasourceParams): Promise<DS> {
-    const dsObj = this.getTemplate(params.templateName, params.startBlock);
+    const dsObj = this.getTemplate(params.templateName, params.startBlock, params.endBlock);
 
     try {
       await this.blockchainService.updateDynamicDs(params, dsObj);

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

@@ -0,0 +1,195 @@
+// Copyright 2020-2025 SubQuery Pte Ltd authors & contributors
+// SPDX-License-Identifier: GPL-3.0
+
+import {BaseCustomDataSource, BaseDataSource} from '@subql/types-core';
+import {IApi} from '../api.service';
+import {IBlockchainService} from '../blockchain.service';
+import {NodeConfig} from '../configure';
+import {ProcessBlockResponse} from './blockDispatcher';
+import {DsProcessorService} from './ds-processor.service';
+import {DatasourceParams, DynamicDsService} from './dynamic-ds.service';
+import {BaseIndexerManager, FilterTypeMap, HandlerInputTypeMap, ProcessorTypeMap} from './indexer.manager';
+import {IndexerSandbox} from './sandbox';
+import {CacheMetadataModel} from './storeModelProvider';
+import {IBlock, ISubqueryProject} from './types';
+import {IUnfinalizedBlocksService} from './unfinalizedBlocks.service';
+
+type FM = FilterTypeMap<BaseDataSource>;
+type PM = ProcessorTypeMap<BaseDataSource, FM>;
+type HIM = HandlerInputTypeMap<BaseDataSource, FM>;
+
+// Minimal sandbox mock that lets us grab the frozen destroy callback
+class MockSandbox {
+  // eslint-disable-next-line @typescript-eslint/ban-types
+  frozenFns: Record<string, Function> = {};
+
+  freeze(value: any, name: string): void {
+    if (typeof value === 'function') {
+      this.frozenFns[name] = value;
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-empty-function
+  async securedExec(): Promise<void> {}
+}
+
+class TestIndexerManager extends BaseIndexerManager<
+  any,
+  any,
+  any,
+  IApi,
+  BaseDataSource,
+  BaseDataSource & BaseCustomDataSource,
+  FM,
+  PM,
+  HIM
+> {
+  processedStartBlocks: number[] = [];
+  destroyConfig?: {triggerStartBlock: number; targetTemplate: string; targetIndex: number};
+
+  async indexBlock(block: IBlock<any>, datasources: BaseDataSource[]): Promise<ProcessBlockResponse> {
+    return this.internalIndexBlock(block, datasources, () => Promise.resolve({} as any));
+  }
+
+  // Simulates what chain-specific indexBlockData does: iterate dataSources, call getVM, run handlers
+  protected async indexBlockData(
+    _block: any,
+    dataSources: BaseDataSource[],
+    getVM: (d: BaseDataSource) => Promise<IndexerSandbox>
+  ): Promise<void> {
+    for (let i = 0; i < dataSources.length; i++) {
+      const ds = dataSources[i];
+      this.processedStartBlocks.push(ds.startBlock!);
+
+      const vm = (await getVM(ds)) as unknown as MockSandbox;
+
+      // Trigger destroy if this ds matches the config
+      if (this.destroyConfig && ds.startBlock === this.destroyConfig.triggerStartBlock) {
+        const destroyFn = vm.frozenFns.destroyDynamicDatasource;
+        if (destroyFn) {
+          await destroyFn(this.destroyConfig.targetTemplate, this.destroyConfig.targetIndex);
+        }
+      }
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  protected async prepareFilteredData<T>(_kind: any, data: T): Promise<T> {
+    return data;
+  }
+}
+
+class TestDynamicDsService extends DynamicDsService<BaseDataSource, ISubqueryProject> {
+  constructor(project: ISubqueryProject) {
+    super(project, {
+      updateDynamicDs: () => Promise.resolve(undefined),
+    } as unknown as IBlockchainService);
+  }
+}
+
+const mockMetadata = (initData: DatasourceParams[] = []) => {
+  let datasourceParams: DatasourceParams[] = initData;
+
+  return {
+    set: (_key: string, value: any) => {
+      datasourceParams = value;
+    },
+    find: (_key: string) => Promise.resolve([...datasourceParams]),
+    setNewDynamicDatasource: (params: DatasourceParams) => datasourceParams.push(params),
+  } as unknown as CacheMetadataModel;
+};
+
+const mockBlock = (height: number): IBlock<any> => ({
+  getHeader: () => ({blockHeight: height, blockHash: `hash-${height}`, parentHash: undefined, timestamp: new Date()}),
+  block: {},
+});
+
+describe('BaseIndexerManager', () => {
+  let dynamicDsService: TestDynamicDsService;
+  let manager: TestIndexerManager;
+
+  const project = {
+    templates: [{name: 'Test'}, {name: 'Other'}],
+  } as any as ISubqueryProject;
+
+  beforeEach(() => {
+    dynamicDsService = new TestDynamicDsService(project);
+
+    manager = new TestIndexerManager(
+      {unsafeApi: {}} as unknown as IApi,
+      {unfinalizedBlocks: false, profiler: false} as unknown as NodeConfig,
+      {getDsProcessor: () => new MockSandbox()} as any,
+      {} as DsProcessorService<BaseDataSource, BaseDataSource & BaseCustomDataSource>,
+      dynamicDsService as any,
+      {processUnfinalizedBlocks: () => Promise.resolve(undefined)} as unknown as IUnfinalizedBlocksService<any>,
+      {} as FM,
+      {} as PM,
+      {isRuntimeDs: () => true, isCustomDs: () => false} as unknown as IBlockchainService
+    );
+  });
+
+  describe('destroy dynamic datasource mid-block', () => {
+    it('removes destroyed ds from the iteration within the same block', async () => {
+      const meta = mockMetadata([
+        {templateName: 'Test', startBlock: 1},
+        {templateName: 'Test', startBlock: 5},
+        {templateName: 'Test', startBlock: 10},
+      ]);
+      await dynamicDsService.init(meta);
+      const datasources = await dynamicDsService.getDynamicDatasources();
+
+      // When processing startBlock=5, destroy ds at index 2 (startBlock=10)
+      manager.destroyConfig = {triggerStartBlock: 5, targetTemplate: 'Test', targetIndex: 2};
+
+      await manager.indexBlock(mockBlock(100), datasources);
+
+      // startBlock=10 should never be reached
+      expect(manager.processedStartBlocks).toEqual([1, 5]);
+    });
+
+    it('sets endBlock on the destroyed datasource', async () => {
+      const meta = mockMetadata([
+        {templateName: 'Test', startBlock: 1},
+        {templateName: 'Test', startBlock: 5},
+      ]);
+      await dynamicDsService.init(meta);
+      const datasources = await dynamicDsService.getDynamicDatasources();
+
+      manager.destroyConfig = {triggerStartBlock: 1, targetTemplate: 'Test', targetIndex: 1};
+
+      await manager.indexBlock(mockBlock(50), datasources);
+
+      const param = dynamicDsService.getDatasourceParamByIndex(1);
+      expect(param?.endBlock).toBe(50);
+    });
+
+    it('processes all datasources when nothing is destroyed', async () => {
+      const meta = mockMetadata([
+        {templateName: 'Test', startBlock: 1},
+        {templateName: 'Test', startBlock: 5},
+        {templateName: 'Test', startBlock: 10},
+      ]);
+      await dynamicDsService.init(meta);
+      const datasources = await dynamicDsService.getDynamicDatasources();
+
+      await manager.indexBlock(mockBlock(100), datasources);
+
+      expect(manager.processedStartBlocks).toEqual([1, 5, 10]);
+    });
+
+    it('filters out already-destroyed datasources before processing starts', async () => {
+      const meta = mockMetadata([
+        {templateName: 'Test', startBlock: 1, endBlock: 50},
+        {templateName: 'Test', startBlock: 5},
+        {templateName: 'Test', startBlock: 10},
+      ]);
+      await dynamicDsService.init(meta);
+      const datasources = await dynamicDsService.getDynamicDatasources();
+
+      await manager.indexBlock(mockBlock(100), datasources);
+
+      // startBlock=1 was destroyed at block 50, should be excluded
+      expect(manager.processedStartBlocks).toEqual([5, 10]);
+    });
+  });
+});

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

@@ -116,6 +116,24 @@ export abstract class BaseIndexerManager<
           dynamicDsCreated = true;
         }, 'createDynamicDatasource');
 
+        // Inject function to get dynamic datasources by template into vm
+        vm.freeze((templateName: string) => {
+          return this.dynamicDsService.getDynamicDatasourcesByTemplate(templateName);
+        }, 'getDynamicDatasources');
+
+        // Inject function to destroy ds into vm
+        vm.freeze(async (templateName: string, index: number) => {
+          await this.dynamicDsService.destroyDynamicDatasource(templateName, blockHeight, index);
+
+          // Re-filter and mutate the array in-place so that indexBlockData's reference
+          // sees the change. A simple reassignment (filteredDataSources = ...) would only
+          // update this closure's variable, leaving the dataSources parameter in
+          // indexBlockData pointing at the stale array.
+          const refiltered = this.filterDataSources(blockHeight, filteredDataSources);
+          filteredDataSources.length = 0;
+          filteredDataSources.push(...refiltered);
+        }, 'destroyDynamicDatasource');
+
         return vm;
       });
     }
@@ -135,11 +153,15 @@ export abstract class BaseIndexerManager<
   private filterDataSources(nextProcessingHeight: number, dataSources: DS[]): DS[] {
     let filteredDs: DS[];
 
+    // Strict `>` (not `>=`) is intentional: destroyDynamicDatasource sets
+    // endBlock = currentBlockHeight, and the destroyed DS must be excluded
+    // when re-filtering within the same block. With `>=` the DS would pass
+    // the filter and remain active for the rest of the block.
     filteredDs = dataSources.filter(
       (ds) =>
         ds.startBlock !== undefined &&
         ds.startBlock <= nextProcessingHeight &&
-        (ds.endBlock ?? Number.MAX_SAFE_INTEGER) >= nextProcessingHeight
+        (ds.endBlock ?? Number.MAX_SAFE_INTEGER) > nextProcessingHeight
     );
 
     // perform filter for custom ds