Sync 2023/07/28

packages/node/package.json

@@ -26,7 +26,7 @@
     "@nestjs/schedule": "^3.0.1",
     "@subql/common": "^2.3.0",
     "@subql/common-algorand": "workspace:*",
-    "@subql/node-core": "^3.1.2",
+    "@subql/node-core": "^4.0.0",
     "@subql/types-algorand": "workspace:*",
     "@willsoto/nestjs-prometheus": "^4.4.0",
     "algosdk": "^2.2.0",

packages/node/src/algorand/algorand.spec.ts

@@ -7,6 +7,7 @@
 import {
   ApiService,
   ConnectionPoolService,
+  ConnectionPoolStateManager,
   NodeConfig,
 } from '@subql/node-core';
 import { GraphQLSchema } from 'graphql';
@@ -45,6 +46,7 @@
         useFactory: () => testSubqueryProject(endpoint, chainId),
       },
       NodeConfig,
+      ConnectionPoolStateManager,
       ConnectionPoolService,
       AlgorandApiService,
     ],
@@ -111,7 +113,7 @@
     };
 
     expect(
       filterTransaction(tx as any, {
         txType: 'acfg',
         sender: '7JMGBIDKQRR4MC3DNC73QU4QUNNN43VNY5RYPN2FRWEG6NXAHQMCPD4BIQ',
       }),

packages/node/src/algorand/api.service.algorand.ts

@@ -62,7 +62,7 @@ export class AlgorandApiService extends ApiService<
           );
         }
 
-        this.connectionPoolService.addToConnections(connection, endpoint);
+        await this.connectionPoolService.addToConnections(connection, endpoint);
       } catch (error) {
         if (retries < maxRetries) {
           retries++;

packages/node/src/configure/configure.module.ts

@@ -59,7 +59,10 @@ function warnDeprecations() {
 @Global()
 @Module({})
 export class ConfigureModule {
-  static async register(): Promise<DynamicModule> {
+  static async getInstance(): Promise<{
+    config: NodeConfig;
+    project: () => Promise<SubqueryProject>;
+  }> {
     const { argv } = yargsOptions;
     let config: NodeConfig;
     let rawManifest: unknown;
@@ -131,6 +134,11 @@ export class ConfigureModule {
       return p;
     };
 
+    return { config, project };
+  }
+  static async register(): Promise<DynamicModule> {
+    const { config, project } = await ConfigureModule.getInstance();
+
     return {
       module: ConfigureModule,
       providers: [

packages/node/src/indexer/api.service.spec.ts

@@ -4,7 +4,11 @@
 import { INestApplication } from '@nestjs/common';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { Test } from '@nestjs/testing';
-import { ConnectionPoolService, NodeConfig } from '@subql/node-core';
+import {
+  ConnectionPoolService,
+  ConnectionPoolStateManager,
+  NodeConfig,
+} from '@subql/node-core';
 import { GraphQLSchema } from 'graphql';
 import { AlgorandApiService } from '../algorand';
 import { SubqueryProject } from '../configure/SubqueryProject';
@@ -43,6 +47,7 @@ describe('ApiService', () => {
           provide: 'ISubqueryProject',
           useFactory: () => testSubqueryProject(endpoint),
         },
+        ConnectionPoolStateManager,
         ConnectionPoolService,
         NodeConfig,
         AlgorandApiService,

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

@@ -17,8 +17,12 @@ import {
   HostDynamicDS,
   WorkerBlockDispatcher,
   IUnfinalizedBlocksService,
+  HostConnectionPoolState,
+  ConnectionPoolStateManager,
+  connectionPoolStateHostFunctions,
 } from '@subql/node-core';
 import { Store } from '@subql/types';
+import { AlgorandApiConnection } from '../../algorand';
 import {
   SubqlProjectDs,
   SubqueryProject,
@@ -37,11 +41,16 @@ async function createIndexerWorker(
   store: Store,
   dynamicDsService: IDynamicDsService<SubqlProjectDs>,
   unfinalizedBlocksService: IUnfinalizedBlocksService<BlockContent>,
+  connectionPoolState: ConnectionPoolStateManager<AlgorandApiConnection>,
   root: string,
 ): Promise<IndexerWorker> {
   const indexerWorker = Worker.create<
     IInitIndexerWorker,
-    HostDynamicDS<SubqlProjectDs> & HostStore & HostUnfinalizedBlocks
+    // HostDynamicDS<SubqlProjectDs> & HostStore & HostUnfinalizedBlocks
+    HostDynamicDS<SubqlProjectDs> &
+      HostStore &
+      HostUnfinalizedBlocks &
+      HostConnectionPoolState<AlgorandApiConnection>
   >(
     path.resolve(__dirname, '../../../dist/indexer/worker/worker.js'),
     [
@@ -71,6 +80,7 @@ async function createIndexerWorker(
         unfinalizedBlocksService.processUnfinalizedBlockHeader.bind(
           unfinalizedBlocksService,
         ),
+      ...connectionPoolStateHostFunctions(connectionPoolState),
     },
     root,
   );
@@ -96,6 +106,7 @@ export class WorkerBlockDispatcherService
     @Inject('ISubqueryProject') project: SubqueryProject,
     dynamicDsService: DynamicDsService,
     unfinalizedBlocksService: UnfinalizedBlocksService,
+    connectionPoolState: ConnectionPoolStateManager<AlgorandApiConnection>,
   ) {
     super(
       nodeConfig,
@@ -112,6 +123,7 @@ export class WorkerBlockDispatcherService
           storeService.getStore(),
           dynamicDsService,
           unfinalizedBlocksService,
+          connectionPoolState,
           project.root,
         ),
     );

packages/node/src/indexer/fetch.module.ts

@@ -15,6 +15,7 @@ import {
   StoreCacheService,
   PgMmrCacheService,
   MmrQueryService,
+  ConnectionPoolStateManager,
 } from '@subql/node-core';
 import { AlgorandApiConnection, AlgorandApiService } from '../algorand';
 import { SubqueryProject } from '../configure/SubqueryProject';
@@ -62,6 +63,7 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
       inject: ['ISubqueryProject', ConnectionPoolService, EventEmitter2],
     },
     IndexerManager,
+    ConnectionPoolStateManager,
     {
       provide: 'IBlockDispatcher',
       useFactory: (
@@ -77,6 +79,7 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
         project: SubqueryProject,
         dynamicDsService: DynamicDsService,
         unfinalizedBlocksService: UnfinalizedBlocksService,
+        connectionPoolState: ConnectionPoolStateManager<AlgorandApiConnection>,
       ) =>
         nodeConfig.workers !== undefined
           ? new WorkerBlockDispatcherService(
@@ -90,6 +93,7 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
               project,
               dynamicDsService,
               unfinalizedBlocksService,
+              connectionPoolState,
             )
           : new BlockDispatcherService(
               apiService,
@@ -117,9 +121,11 @@ import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
         'ISubqueryProject',
         DynamicDsService,
         UnfinalizedBlocksService,
+        ConnectionPoolStateManager,
       ],
     },
     FetchService,
+    ConnectionPoolService,
     IndexingBenchmarkService,
     PoiBenchmarkService,
     DictionaryService,

packages/node/src/indexer/indexer.module.ts

@@ -13,6 +13,8 @@ import {
   WorkerDynamicDsService,
   PgMmrCacheService,
   MmrQueryService,
+  ConnectionPoolStateManager,
+  WorkerConnectionPoolStateManager,
 } from '@subql/node-core';
 import { AlgorandApiService, AlgorandApiConnection } from '../algorand';
 import { SubqueryProject } from '../configure/SubqueryProject';
@@ -29,6 +31,15 @@ import { WorkerService } from './worker/worker.service';
     IndexerManager,
     StoreCacheService,
     StoreService,
+    {
+      provide: ConnectionPoolStateManager,
+      useFactory: () => {
+        if (isMainThread) {
+          throw new Error('Expected to be worker thread');
+        }
+        return new WorkerConnectionPoolStateManager((global as any).host);
+      },
+    },
     ConnectionPoolService,
     UnfinalizedBlocksService,
     {

packages/node/src/indexer/worker/worker.ts

@@ -31,7 +31,10 @@ import {
   hostDynamicDsKeys,
   HostDynamicDS,
   ProcessBlockResponse,
+  HostConnectionPoolState,
+  hostConnectionPoolStateKeys,
 } from '@subql/node-core';
+import { AlgorandApiConnection } from '../../algorand';
 import { SubqlProjectDs } from '../../configure/SubqueryProject';
 import { IndexerManager } from '../indexer.manager';
 import { WorkerModule } from './worker.module';
@@ -40,6 +43,10 @@ import {
   WorkerService,
   WorkerStatusResponse,
 } from './worker.service';
+import {
+  HostUnfinalizedBlocks,
+  hostUnfinalizedBlocksKeys,
+} from './worker.unfinalizedBlocks.service';
 let app: INestApplication;
 let workerService: WorkerService;
 
@@ -116,10 +123,18 @@ async function waitForWorkerBatchSize(heapSizeInBytes: number): Promise<void> {
 
 // Register these functions to be exposed to worker host
 (global as any).host = WorkerHost.create<
-  HostStore & HostDynamicDS<SubqlProjectDs>,
+  HostStore &
+    HostDynamicDS<SubqlProjectDs> &
+    HostUnfinalizedBlocks &
+    HostConnectionPoolState<AlgorandApiConnection>,
   IInitIndexerWorker
 >(
-  [...hostStoreKeys, ...hostDynamicDsKeys],
+  [
+    ...hostStoreKeys,
+    ...hostDynamicDsKeys,
+    ...hostUnfinalizedBlocksKeys,
+    ...hostConnectionPoolStateKeys,
+  ],
   {
     initWorker,
     fetchBlock,

packages/node/src/init.ts

@@ -48,10 +48,7 @@ export async function bootstrap(): Promise<void> {
 
     const projectService = app.get('IProjectService');
     const fetchService = app.get(FetchService);
-    const apiService = app.get(AlgorandApiService);
 
-    // Initialise async services, we do this here rather than in factories, so we can capture one off events
-    await apiService.init();
     await projectService.init();
     await fetchService.init(projectService.startHeight);
 

packages/node/src/subcommands/testing.init.ts

@@ -1,29 +1,17 @@
 // Copyright 2020-2023 SubQuery Pte Ltd authors & contributors
 // SPDX-License-Identifier: GPL-3.0
 
-import { NestFactory } from '@nestjs/core';
-import { getLogger, NestLogger } from '@subql/node-core';
-import { AlgorandApiService } from '../algorand';
-import { ProjectService } from '../indexer/project.service';
-import { TestingModule } from './testing.module';
+import { getLogger } from '@subql/node-core';
+import { ConfigureModule } from '../configure/configure.module';
 import { TestingService } from './testing.service';
 
 const logger = getLogger('Testing');
 export async function testingInit(): Promise<void> {
   try {
-    const app = await NestFactory.create(TestingModule, {
-      logger: new NestLogger(),
-    });
+    const { config, project } = await ConfigureModule.getInstance();
+    const subqueryProject = await project();
 
-    await app.init();
-    const projectService = app.get(ProjectService);
-    const apiService = app.get(AlgorandApiService);
-
-    // Initialise async services, we do this here rather than in factories, so we can capture one off events
-    await apiService.init();
-    await projectService.init();
-
-    const testingService = app.get(TestingService);
+    const testingService = new TestingService(config, subqueryProject);
     await testingService.init();
     await testingService.run();
   } catch (e) {

packages/node/src/subcommands/testing.module.ts

@@ -10,6 +10,7 @@ import {
   NodeConfig,
   PoiService,
   StoreService,
+  TestRunner,
 } from '@subql/node-core';
 import { AlgorandApiConnection, AlgorandApiService } from '../algorand';
 import { ConfigureModule } from '../configure/configure.module';
@@ -20,6 +21,7 @@ import { FetchModule } from '../indexer/fetch.module';
 import { IndexerManager } from '../indexer/indexer.manager';
 import { ProjectService } from '../indexer/project.service';
 import { SandboxService } from '../indexer/sandbox.service';
+import { UnfinalizedBlocksService } from '../indexer/unfinalizedBlocks.service';
 import { MetaModule } from '../meta/meta.module';
 import { TestingService } from './testing.service';
 
@@ -33,6 +35,7 @@ import { TestingService } from './testing.service';
     DsProcessorService,
     DynamicDsService,
     ProjectService,
+    UnfinalizedBlocksService,
     ConnectionPoolService,
     {
       provide: 'IProjectService',
@@ -57,10 +60,20 @@ import { TestingService } from './testing.service';
     },
     IndexerManager,
     SchedulerRegistry,
+    TestRunner,
+    {
+      provide: 'IApi',
+      useClass: AlgorandApiService,
+    },
+    {
+      provide: 'IIndexerManager',
+      useClass: IndexerManager,
+    },
   ],
 
   imports: [MetaModule, FetchModule],
   controllers: [],
+  exports: [TestRunner],
 })
 export class TestingFeatureModule {}