Development to main (#1546)

* Enhance error handling in GatewayProxyController for block retrieval by hash (#1545)

* use EsdtType instead of hardcoded values (#1544)

* token service improvements (#1543)

* processes 50 tokens concurrently using ConcurrencyUtils

* run all three batch operations in parallel with increased concurrency

* add more logs

* apply mex prices

* add logs ( to be removed )

* increse cron time expression

* set cron to 2 minutes

* Websocket subscriptions (#1528)

* add websockets for blocks and txs

* add support for subscribe to stats

* use websockets rooms

* remove logs

* add lock on crons

* check stats room exists

Signed-off-by: GuticaStefan <stefan.gutica@gmail.com>

* fix indent spaces

* add validation pipes + filters

* add try catch + class validator fixes

* fix linter

* add pool subscription + reduce filters combinations for subscriptions

* lint

* add support for events subscription

* lint

* separate subscription websocket into separate app

* add config

* add path

* fix

* add path for events + config default settings

* temp logs

* temp logs 2

* added missing configs + remove temp logs

* enable andromeda in config

* add metrics on subscription

* remove async + reschedule

* refresh metrics every second

* set max listeners to 12

* add EOL

* add count on update + parallel broadcast to rooms

* lower ttl for blocks count cache

* remove comments

* remove comments

* renaming

---------

Signed-off-by: GuticaStefan <stefan.gutica@gmail.com>
Co-authored-by: bogdan-rosianu <bogdan.rosianu@yahoo.com>
Co-authored-by: cfaur09 <catalinfaurpaul@gmail.com>

* log instead of error for invalid legacy delegation contract (#1542)

---------

Signed-off-by: GuticaStefan <stefan.gutica@gmail.com>
Co-authored-by: Catalin Faur <52102171+cfaur09@users.noreply.github.com>
Co-authored-by: Gutica Stefan <123564494+stefangutica@users.noreply.github.com>
Co-authored-by: cfaur09 <catalinfaurpaul@gmail.com>

Author: 3 people

config/config.devnet.yaml

@@ -8,7 +8,7 @@ api:
   websocket: true
 cron:
   cacheWarmer: true
-  fastWarm: true
+  fastWarm: false
   queueWorker: true
   elasticUpdater: false
 flags:
@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -65,7 +68,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1043
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

config/config.e2e-mocked.mainnet.yaml

@@ -5,6 +5,9 @@ api:
   private: true
   graphql: true
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   dataApi:
     enabled: false
     serviceUrl: 'https://data-api.multiversx.com'

config/config.e2e.mainnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -63,7 +66,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1391
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

config/config.mainnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674

config/config.testnet.yaml

@@ -20,6 +20,9 @@ flags:
   processNfts: true
   collectionPropertiesFromGateway: false
 features:
+  websocketSubscription:
+    enabled: false
+    port: 6002
   eventsNotifier:
     enabled: false
     port: 5674
@@ -62,7 +65,7 @@ features:
     cronExpression: '*/5 * * * * *'
     activationEpoch: 1043
   chainAndromeda:
-    enabled: false
+    enabled: true
     activationEpoch: 4
   nodeEpochsLeft:
     enabled: false

package-lock.json

@@ -113,11 +113,13 @@
     "@sendgrid/mail": "^8.1.5",
     "agentkeepalive": "^4.2.1",
     "amqp-connection-manager": "^4.1.3",
-    "anchorme": "^3.0.8",
     "amqplib": "^0.10.0",
+    "anchorme": "^3.0.8",
     "apollo-server-core": "^3.13.0",
     "apollo-server-express": "3.13.0",
     "bignumber.js": "^9.0.2",
+    "class-transformer": "^0.5.1",
+    "class-validator": "^0.14.2",
     "compression": "^1.8.0",
     "crypto-js": "^4.1.1",
     "dataloader": "^2.2.2",
@@ -144,6 +146,8 @@
     "rxjs": "^7.1.0",
     "sharp": "^0.34.2",
     "simple-git": "^3.16.0",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "swagger-ui-express": "^4.3.0",
     "tiny-async-pool": "^1.2.0",
     "typeorm": "^0.3.25",
@@ -216,4 +220,4 @@
       "node_modules"
     ]
   }
-}
+}

package.json

@@ -954,4 +954,22 @@ export class ApiConfigService {
   getCompressionChunkSize(): number {
     return this.configService.get<number>('compression.chunkSize') ?? 16384;
   }
+
+  getIsWebsocketSubscriptionActive(): boolean {
+    const isWebsocketSubscriptionActive = this.configService.get<boolean>('features.websocketSubscription.enabled');
+    if (isWebsocketSubscriptionActive === undefined) {
+      throw new Error('No features.websocketSubscription.enabled flag present');
+    }
+
+    return isWebsocketSubscriptionActive;
+  }
+
+  getWebsocketSubscriptionPort(): number {
+    const port = this.configService.get<number>('features.websocketSubscription.port');
+    if (port === undefined) {
+      throw new Error('No features.websocketSubscription.port present');
+    }
+
+    return port;
+  }
 }

src/common/api-config/api.config.service.ts

@@ -22,6 +22,7 @@ export class ApiMetricsService {
   private static transactionsCompletedCounter: Counter<string>;
   private static transactionsPendingResultsCounter: Counter<string>;
   private static batchUpdatesCounter: Counter<string>;
+  private static subscriptionsConnectionsGauge: Gauge<string>;
 
   constructor(
     private readonly apiConfigService: ApiConfigService,
@@ -32,6 +33,13 @@ export class ApiMetricsService {
     private readonly metricsService: MetricsService,
   ) {
 
+    if (!ApiMetricsService.subscriptionsConnectionsGauge) {
+      ApiMetricsService.subscriptionsConnectionsGauge = new Gauge({
+        name: 'websocket_subscriptions_connections',
+        help: 'Number of websocket connections for subscriptions',
+      });
+    }
+
     if (!ApiMetricsService.vmQueriesHistogram) {
       ApiMetricsService.vmQueriesHistogram = new Histogram({
         name: 'vm_query',
@@ -182,6 +190,14 @@ export class ApiMetricsService {
     ApiMetricsService.lastProcessedTransactionCompletedProcessorNonce.set({ shardId }, nonce);
   }
 
+  @OnEvent(MetricsEvents.SetWebsocketMetrics)
+  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number }) {
+    const { connectedClients } = payload;
+
+    ApiMetricsService.subscriptionsConnectionsGauge.set(connectedClients);
+  }
+
+
   @OnEvent(MetricsEvents.SetTransactionsCompleted)
   recordTransactionsCompleted(payload: { transactions: any[] }) {
     ApiMetricsService.transactionsCompletedCounter.inc(payload.transactions.length);

src/common/metrics/api.metrics.service.ts

@@ -156,7 +156,7 @@ export class CacheWarmerService {
     await this.invalidateKey(CacheInfo.TransactionPool.key, pool, this.apiConfigService.getTransactionPoolCacheWarmerTtlInSeconds());
   }
 
-  @Cron(CronExpression.EVERY_MINUTE)
+  @Cron('*/2 * * * *')
   @Lock({ name: 'All Tokens invalidations', verbose: true })
   async handleEsdtTokenInvalidations() {
     const tokens = await this.tokenService.getAllTokensRaw();