Disable cache

packages/node-core/src/configure/ProjectUpgrade.service.spec.ts

@@ -4,6 +4,7 @@
 import {SchedulerRegistry} from '@nestjs/schedule';
 import {Sequelize} from '@subql/x-sequelize';
 import {CacheMetadataModel, ISubqueryProject, StoreCacheService, StoreService} from '../indexer';
+import {IStoreModelService} from '../indexer/storeCache/storeModel.service';
 import {NodeConfig} from './NodeConfig';
 import {IProjectUpgradeService, ProjectUpgradeService, upgradableSubqueryProject} from './ProjectUpgrade.service';
 
@@ -289,7 +290,7 @@ describe('Project Upgrades', () => {
   describe('Upgradable subquery project', () => {
     let upgradeService: ProjectUpgradeService<ISubqueryProject>;
     let project: ISubqueryProject & IProjectUpgradeService<ISubqueryProject>;
-    let storeCache: StoreCacheService;
+    let storeCache: IStoreModelService;
 
     beforeEach(async () => {
       storeCache = new StoreCacheService({} as any, {} as any, {} as any, new SchedulerRegistry());

packages/node-core/src/configure/ProjectUpgrade.service.ts

@@ -7,7 +7,14 @@ import {ParentProject} from '@subql/types-core';
 import {Sequelize, Transaction} from '@subql/x-sequelize';
 import {findLast, last, parseInt} from 'lodash';
 import {SchemaMigrationService} from '../db';
-import {CacheMetadataModel, ISubqueryProject, StoreCacheService, StoreService} from '../indexer';
+import {
+  CacheMetadataModel,
+  IMetadata,
+  IStoreModelService,
+  ISubqueryProject,
+  StoreCacheService,
+  StoreService,
+} from '../indexer';
 import {getLogger} from '../logger';
 import {exitWithError, monitorWrite} from '../process';
 import {getStartHeight, mainThreadOnly} from '../utils';
@@ -107,7 +114,7 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
   #currentHeight: number;
   #currentProject: P;
 
-  #storeCache?: StoreCacheService;
+  #storeCache?: IStoreModelService;
   #initialized = false;
 
   private config?: NodeConfig;
@@ -150,16 +157,10 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
       return;
     }
     this.#initialized = true;
-    this.#storeCache = storeService.storeCache;
+    this.#storeCache = storeService.storeModel;
     this.config = config;
 
-    this.migrationService = new SchemaMigrationService(
-      sequelize,
-      storeService,
-      storeService.storeCache._flushCache.bind(storeService.storeCache),
-      schema,
-      config
-    );
+    this.migrationService = new SchemaMigrationService(sequelize, storeService, schema, config);
 
     const indexedDeployments = await this.getDeploymentsMetadata();
 
@@ -172,7 +173,9 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
     } catch (e: any) {
       if (e instanceof EntryNotFoundError) {
         throw new Error(
-          `Unable to find project for height ${this.#currentHeight}. If the project start height is increased it will not jump to that block. Please either reindex or specify blocks to bypass.`,
+          `Unable to find project for height ${
+            this.#currentHeight
+          }. If the project start height is increased it will not jump to that block. Please either reindex or specify blocks to bypass.`,
           {cause: e}
         );
       }
@@ -207,7 +210,7 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
     return this.#currentHeight;
   }
 
-  private get metadata(): CacheMetadataModel {
+  private get metadata(): IMetadata {
     assert(this.#storeCache?.metadata, 'Project Upgrades service has not been initialized, unable to update metadata');
     return this.#storeCache.metadata;
   }
@@ -243,7 +246,7 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
       nextId = iterator.next();
     }
     // this remove any deployments in metadata beyond target height
-    await this.removeIndexedDeployments(targetBlockHeight);
+    await this.removeIndexedDeployments(targetBlockHeight, transaction);
   }
 
   private async migrate(
@@ -256,7 +259,7 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
       assert(this.migrationService, 'MigrationService is undefined');
       if (this.config.allowSchemaMigration) {
         await this.migrationService.run(project.schema, newProject.schema, transaction);
-        this.metadata.setIncrement('schemaMigrationCount');
+        await this.metadata.setIncrement('schemaMigrationCount', undefined, transaction);
       }
     }
   }
@@ -450,11 +453,11 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
 
     deployments[blockHeight] = id;
 
-    this.metadata.set('deployments', JSON.stringify(deployments));
+    await this.metadata.set('deployments', JSON.stringify(deployments));
   }
 
   // Remove metadata deployments beyond this blockHeight
-  async removeIndexedDeployments(blockHeight: number): Promise<void> {
+  async removeIndexedDeployments(blockHeight: number, tx?: Transaction): Promise<void> {
     const deployments = await this.getDeploymentsMetadata();
 
     // remove all future block heights
@@ -466,6 +469,6 @@ export class ProjectUpgradeService<P extends ISubqueryProject = ISubqueryProject
       }
     });
 
-    this.metadata.set('deployments', JSON.stringify(deployments));
+    await this.metadata.set('deployments', JSON.stringify(deployments), tx);
   }
 }

packages/node-core/src/db/migration-service/SchemaMigration.service.test.ts

@@ -46,13 +46,7 @@ async function setup(
   await storeService.initCoreTables(schemaName);
   await initDbSchema(schemaName, storeService);
 
-  return new SchemaMigrationService(
-    sequelize,
-    storeService,
-    storeCache._flushCache.bind(storeCache),
-    schemaName,
-    config
-  );
+  return new SchemaMigrationService(sequelize, storeService, schemaName, config);
 }
 
 function loadGqlSchema(fileName: string): GraphQLSchema {

packages/node-core/src/db/migration-service/SchemaMigration.service.ts

@@ -29,7 +29,6 @@ export class SchemaMigrationService {
   constructor(
     private sequelize: Sequelize,
     private storeService: StoreService,
-    private flushCache: (flushAll?: boolean) => Promise<void>,
     private dbSchema: string,
     private config: NodeConfig,
     private dbType: SUPPORT_DB = SUPPORT_DB.postgres
@@ -115,7 +114,7 @@ export class SchemaMigrationService {
     const sortedAddedModels = alignModelOrder<GraphQLModelsType[]>(sortedSchemaModels, addedModels);
     const sortedModifiedModels = alignModelOrder<ModifiedModels>(sortedSchemaModels, modifiedModels);
 
-    await this.flushCache(true);
+    await this.storeService.storeModel.flushData?.(true);
     const migrationAction = await Migration.create(
       this.sequelize,
       this.storeService,
@@ -184,10 +183,10 @@ export class SchemaMigrationService {
       const modelChanges = await migrationAction.run(transaction);
 
       // Update any relevant application state so the right models are used
-      this.storeService.storeCache.updateModels(modelChanges);
+      this.storeService.storeModel.updateModels(modelChanges);
       await this.storeService.updateModels(this.dbSchema, getAllEntitiesRelations(nextSchema));
 
-      await this.flushCache();
+      await this.storeService.storeModel.flushData?.(true);
     } catch (e: any) {
       logger.error(e, 'Failed to execute Schema Migration');
       throw e;

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

@@ -5,17 +5,18 @@ import assert from 'assert';
 
 import {EventEmitter2, OnEvent} from '@nestjs/event-emitter';
 import {hexToU8a, u8aEq} from '@subql/utils';
+import {Transaction} from '@subql/x-sequelize';
 import {NodeConfig, IProjectUpgradeService} from '../../configure';
 import {AdminEvent, IndexerEvent, PoiEvent, TargetBlockPayload} from '../../events';
 import {getLogger} from '../../logger';
-import {exitWithError, monitorCreateBlockFork, monitorCreateBlockStart, monitorWrite} from '../../process';
+import {monitorCreateBlockFork, monitorCreateBlockStart, monitorWrite} from '../../process';
 import {IQueue, mainThreadOnly} from '../../utils';
 import {MonitorServiceInterface} from '../monitor.service';
 import {PoiBlock, PoiSyncService} from '../poi';
 import {SmartBatchService} from '../smartBatch.service';
 import {StoreService} from '../store.service';
-import {StoreCacheService} from '../storeCache';
-import {CachePoiModel} from '../storeCache/cachePoi';
+import {IStoreModelService} from '../storeCache';
+import {IPoi} from '../storeCache/poi';
 import {IBlock, IProjectService, ISubqueryProject} from '../types';
 
 const logger = getLogger('BaseBlockDispatcherService');
@@ -63,7 +64,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
     private projectUpgradeService: IProjectUpgradeService,
     protected queue: Q,
     protected storeService: StoreService,
-    private storeCacheService: StoreCacheService,
+    private storeModelService: IStoreModelService,
     private poiSyncService: PoiSyncService,
     protected monitorService?: MonitorServiceInterface
   ) {
@@ -75,7 +76,7 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
   async init(onDynamicDsCreated: (height: number) => void): Promise<void> {
     this._onDynamicDsCreated = onDynamicDsCreated;
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    this.setProcessedBlockCount((await this.storeCacheService.metadata.find('processedBlockCount', 0))!);
+    this.setProcessedBlockCount((await this.storeModelService.metadata.find('processedBlockCount', 0))!);
   }
 
   get queueSize(): number {
@@ -198,10 +199,10 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
         throw e;
       }
     } else {
-      this.updateStoreMetadata(height);
+      await this.updateStoreMetadata(height, undefined, this.storeService.transaction);
 
       const operationHash = this.storeService.getOperationMerkleRoot();
-      this.createPOI(height, blockHash, operationHash);
+      await this.createPOI(height, blockHash, operationHash, this.storeService.transaction);
 
       if (dynamicDsCreated) {
         this.onDynamicDsCreated(height);
@@ -215,21 +216,38 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
       this.setLatestProcessedHeight(height);
     }
 
-    if (this.nodeConfig.storeCacheAsync) {
-      // Flush all completed block data and don't wait
-      await this.storeCacheService.flushAndWaitForCapacity(false)?.catch((e) => {
-        exitWithError(new Error(`Flushing cache failed`, {cause: e}), logger);
-      });
-    } else {
-      // Flush all data from cache and wait
-      await this.storeCacheService.flushCache(false);
-    }
-
-    if (!this.projectService.hasDataSourcesAfterHeight(height)) {
-      const msg = `All data sources have been processed up to block number ${height}. Exiting gracefully...`;
-      await this.storeCacheService.flushCache(false);
-      exitWithError(msg, logger, 0);
-    }
+    await this.storeModelService.applyPendingChanges(height, !this.projectService.hasDataSourcesAfterHeight(height));
+
+    // if (this.storeModelService instanceof StorCeacheService) {
+    //   if (this.nodeConfig.storeCacheAsync) {
+    //     // Flush all completed block data and don't wait
+    //     await this.storeModelService.flushAndWaitForCapacity(false)?.catch((e) => {
+    //       exitWithError(new Error(`Flushing cache failed`, { cause: e }), logger);
+    //     });
+    //   } else {
+    //     // Flush all data from cache and wait
+    //     await this.storeModelService.flushCache(false);
+    //   }
+
+    //   if (!this.projectService.hasDataSourcesAfterHeight(height)) {
+    //     const msg = `All data sources have been processed up to block number ${height}. Exiting gracefully...`;
+    //     await this.storeModelService.flushCache(false);
+    //     exitWithError(msg, logger, 0);
+    //   }
+    // } else if (this.storeModelService instanceof PlainStoreModelService) {
+    //   const tx = this.storeService.transaction;
+    //   if (!tx) {
+    //     exitWithError(new Error('Transaction not found'), logger, 1);
+    //   }
+    //   await tx.commit();
+
+    //   if (!this.projectService.hasDataSourcesAfterHeight(height)) {
+    //     const msg = `All data sources have been processed up to block number ${height}. Exiting gracefully...`;
+    //     exitWithError(msg, logger, 0);
+    //   }
+    // } else {
+    //   exitWithError(new Error('Unknown store model service'), logger, 1);
+    // }
   }
 
   @OnEvent(AdminEvent.rewindTarget)
@@ -258,7 +276,12 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
    * @param operationHash
    * @private
    */
-  private createPOI(height: number, blockHash: string, operationHash: Uint8Array): void {
+  private async createPOI(
+    height: number,
+    blockHash: string,
+    operationHash: Uint8Array,
+    tx?: Transaction
+  ): Promise<void> {
     if (!this.nodeConfig.proofOfIndex) {
       return;
     }
@@ -268,30 +291,33 @@ export abstract class BaseBlockDispatcher<Q extends IQueue, DS, B> implements IB
 
     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}]);
+    await this.poi.bulkUpsert([poiBlock], tx);
+    await this.storeModelService.metadata.setBulk([{key: 'lastCreatedPoiHeight', value: height}], tx);
     this.eventEmitter.emit(PoiEvent.PoiTarget, {
       height,
       timestamp: Date.now(),
     });
   }
 
   @mainThreadOnly()
-  private updateStoreMetadata(height: number, updateProcessed = true): void {
-    const meta = this.storeCacheService.metadata;
+  private async updateStoreMetadata(height: number, updateProcessed = true, tx?: Transaction): Promise<void> {
+    const meta = this.storeModelService.metadata;
     // Update store metadata
-    meta.setBulk([
-      {key: 'lastProcessedHeight', value: height},
-      {key: 'lastProcessedTimestamp', value: Date.now()},
-    ]);
+    await meta.setBulk(
+      [
+        {key: 'lastProcessedHeight', value: height},
+        {key: 'lastProcessedTimestamp', value: Date.now()},
+      ],
+      tx
+    );
     // Db Metadata increase BlockCount, in memory ref to block-dispatcher _processedBlockCount
     if (updateProcessed) {
-      meta.setIncrement('processedBlockCount');
+      await meta.setIncrement('processedBlockCount', undefined, tx);
     }
   }
 
-  private get poi(): CachePoiModel {
-    const poi = this.storeCacheService.poi;
+  private get poi(): IPoi {
+    const poi = this.storeModelService.poi;
     if (!poi) {
       throw new Error('Poi service expected poi repo but it was not found');
     }

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

@@ -15,7 +15,7 @@ import {monitorWrite} from '../../process';
 import {AutoQueue, isTaskFlushedError} from '../../utils';
 import {MonitorServiceInterface} from '../monitor.service';
 import {StoreService} from '../store.service';
-import {StoreCacheService} from '../storeCache';
+import {IStoreModelService, StoreCacheService} from '../storeCache';
 import {ISubqueryProject, IProjectService} from '../types';
 import {isBlockUnavailableError} from '../worker/utils';
 import {BaseBlockDispatcher} from './base-block-dispatcher';
@@ -58,7 +58,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
     projectService: IProjectService<DS>,
     projectUpgradeService: IProjectUpgradeService,
     storeService: StoreService,
-    storeCacheService: StoreCacheService,
+    storeModelService: IStoreModelService,
     poiSyncService: PoiSyncService,
     project: ISubqueryProject,
     private createIndexerWorker: () => Promise<W>,
@@ -72,7 +72,7 @@ export abstract class WorkerBlockDispatcher<DS, W extends Worker, B>
       projectUpgradeService,
       initAutoQueue(nodeConfig.workers, nodeConfig.batchSize, nodeConfig.timeout, 'Worker'),
       storeService,
-      storeCacheService,
+      storeModelService,
       poiSyncService,
       monitorService
     );

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

@@ -69,7 +69,7 @@ describe('DynamicDsService', () => {
     const meta = mockMetadata([testParam1, testParam2, testParam3, testParam4]);
     await service.init(meta);
 
-    await service.resetDynamicDatasource(2);
+    await service.resetDynamicDatasource(2, null as any);
 
     await expect(meta.find('dynamicDatasources')).resolves.toEqual([testParam1, testParam2]);
     await expect(service.getDynamicDatasources()).resolves.toEqual([testParam1, testParam2]);
@@ -79,7 +79,7 @@ describe('DynamicDsService', () => {
     const meta = mockMetadata([testParam1, testParam2]);
     await service.init(meta);
 
-    meta.set('dynamicDatasources', [testParam1, testParam2, testParam3, testParam4]);
+    await meta.set('dynamicDatasources', [testParam1, testParam2, testParam3, testParam4]);
 
     await expect(service.getDynamicDatasources()).resolves.toEqual([testParam1, testParam2]);
     await expect(service.getDynamicDatasources(true)).resolves.toEqual([