Websocket subscriptions improvements

config/config.devnet.yaml

@@ -25,7 +25,8 @@ features:
     port: 6002
     maxSubscriptionsPerInstance: 10000
     maxSubscriptionsPerClient: 10
-    broadcastIntervalMs: 1000
+    broadcastIntervalMs: 600
+    compressionThreshold: 1024
   eventsNotifier:
     enabled: false
     port: 5674

config/config.e2e-mocked.mainnet.yaml

@@ -11,6 +11,7 @@ features:
     maxSubscriptionsPerInstance: 10000
     maxSubscriptionsPerClient: 10
     broadcastIntervalMs: 600
+    compressionThreshold: 1024
   dataApi:
     enabled: false
     serviceUrl: 'https://data-api.multiversx.com'

config/config.e2e.mainnet.yaml

@@ -26,6 +26,7 @@ features:
     maxSubscriptionsPerInstance: 10000
     maxSubscriptionsPerClient: 10
     broadcastIntervalMs: 600
+    compressionThreshold: 1024
   eventsNotifier:
     enabled: false
     port: 5674

config/config.mainnet.yaml

@@ -26,6 +26,7 @@ features:
     maxSubscriptionsPerInstance: 10000
     maxSubscriptionsPerClient: 10
     broadcastIntervalMs: 600
+    compressionThreshold: 1024
   eventsNotifier:
     enabled: false
     port: 5674

config/config.testnet.yaml

@@ -25,7 +25,8 @@ features:
     port: 6002
     maxSubscriptionsPerInstance: 10000
     maxSubscriptionsPerClient: 10
-    broadcastIntervalMs: 1000
+    broadcastIntervalMs: 600
+    compressionThreshold: 1024
   eventsNotifier:
     enabled: false
     port: 5674

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

@@ -1016,6 +1016,10 @@ export class ApiConfigService {
     return this.configService.get<number>('features.websocketSubscription.broadcastIntervalMs') ?? 1000;
   }
 
+  getWebsocketSubscriptionCompressionThreshold(): number | undefined {
+    return this.configService.get<number>('features.websocketSubscription.compressionThreshold');
+  }
+
   getWebsocketMaxSubscriptionsPerInstance(): number {
     return this.configService.get<number>('features.websocketSubscription.maxSubscriptionsPerInstance') ?? 10_000;
   }

src/common/websockets/subscription-socket-adapter.ts

@@ -0,0 +1,24 @@
+import { INestApplicationContext } from '@nestjs/common';
+import { IoAdapter } from '@nestjs/platform-socket.io';
+import { ServerOptions } from 'socket.io';
+
+export class SubscriptionSocketAdapter extends IoAdapter {
+  constructor(
+    app: INestApplicationContext,
+    private readonly compressionThreshold: number | undefined,
+  ) {
+    super(app);
+  }
+
+  createIOServer(port: number, options?: ServerOptions & { namespace?: string, server?: any }) {
+    if (this.compressionThreshold === undefined) {
+      return super.createIOServer(port, options);
+    }
+
+    // payloads below the threshold are sent as they are, compressing them costs more than it saves
+    return super.createIOServer(port, {
+      ...options,
+      perMessageDeflate: { threshold: this.compressionThreshold },
+    });
+  }
+}

src/crons/websocket/blocks.gateway.ts

@@ -8,6 +8,7 @@ import { UseFilters, UseInterceptors } from '@nestjs/common';
 import { WebsocketExceptionsFilter } from 'src/utils/ws-exceptions.filter';
 import { WsValidationPipe } from 'src/utils/ws-validation.pipe';
 import { OriginLogger } from '@multiversx/sdk-nestjs-common';
+import { RoomKeyGenerator } from './room.key.generator';
 import { LockingGuardInterceptor } from 'src/utils/locking.guard.interceptor';
 
 @UseFilters(WebsocketExceptionsFilter)
@@ -27,7 +28,7 @@ export class BlocksGateway {
     @ConnectedSocket() client: Socket,
     @MessageBody(new WsValidationPipe()) payload: BlockSubscribePayload
   ) {
-    const filterIdentifier = JSON.stringify(payload);
+    const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
     const roomName = `${BlocksGateway.keyPrefix}${filterIdentifier}`;
 
     if (!client.rooms.has(roomName)) {
@@ -42,7 +43,7 @@ export class BlocksGateway {
     @ConnectedSocket() client: Socket,
     @MessageBody(new WsValidationPipe()) payload: BlockSubscribePayload
   ) {
-    const filterIdentifier = JSON.stringify(payload);
+    const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
     const roomName = `${BlocksGateway.keyPrefix}${filterIdentifier}`;
 
     if (client.rooms.has(roomName)) {

src/crons/websocket/custom.subscriptions.data.fetcher.ts

@@ -0,0 +1,128 @@
+import { Injectable } from '@nestjs/common';
+import { OriginLogger } from '@multiversx/sdk-nestjs-common';
+import { QueryPagination } from 'src/common/entities/query.pagination';
+import { Transaction } from 'src/endpoints/transactions/entities/transaction';
+import { TransactionDetailed } from 'src/endpoints/transactions/entities/transaction.detailed';
+import { TransactionFilter } from 'src/endpoints/transactions/entities/transaction.filter';
+import { TransactionQueryOptions } from 'src/endpoints/transactions/entities/transactions.query.options';
+import { TransactionType } from 'src/endpoints/transactions/entities/transaction.type';
+import { TransferService } from 'src/endpoints/transfers/transfer.service';
+import { EventsService } from 'src/endpoints/events/events.service';
+import { EventsFilter } from 'src/endpoints/events/entities/events.filter';
+import { Events } from 'src/endpoints/events/entities/events';
+
+export class CustomSubscriptionsRoundData {
+  constructor(init?: Partial<CustomSubscriptionsRoundData>) {
+    Object.assign(this, init);
+  }
+
+  transactions: Transaction[] = [];
+  transfers: Transaction[] = [];
+  events: Events[] = [];
+}
+
+@Injectable()
+export class CustomSubscriptionsDataFetcher {
+  private readonly logger = new OriginLogger(CustomSubscriptionsDataFetcher.name);
+
+  private static readonly batchSize = 10000;
+
+  constructor(
+    private readonly transferService: TransferService,
+    private readonly eventsService: EventsService,
+  ) { }
+
+  // Fetches everything the custom subscription gateways need for a given round, once.
+  // The 'operations' index holds both transactions and smart contract results, so the
+  // transactions payload is derived from the transfers payload instead of being queried again.
+  async fetchRoundData(timestampMs: number): Promise<CustomSubscriptionsRoundData> {
+    const [transfers, events] = await Promise.all([
+      this.fetchTransfers(timestampMs),
+      this.fetchEvents(timestampMs),
+    ]);
+
+    return new CustomSubscriptionsRoundData({
+      transfers,
+      transactions: this.extractTransactions(transfers),
+      events,
+    });
+  }
+
+  private async fetchTransfers(timestampMs: number): Promise<Transaction[]> {
+    try {
+      const size = CustomSubscriptionsDataFetcher.batchSize;
+      const filter = new TransactionFilter({ before: timestampMs, after: timestampMs, withTxsRelayedByAddress: true });
+      const options = new TransactionQueryOptions({ withScamInfo: false, withUsername: true, withBlockInfo: false, withLogs: false, withOperations: false, withActionTransferValue: false, withTxsOrder: false, withCanBeIgnoredFlag: true });
+
+      const allTransfers: Transaction[] = [];
+
+      let batch = await this.transferService.getTransfers(filter, new QueryPagination({ size }), options);
+      allTransfers.push(...batch);
+
+      while (batch.length === size) {
+        const searchAfter = batch[batch.length - 1].searchAfter;
+        if (searchAfter == null) {
+          break;
+        }
+
+        batch = await this.transferService.getTransfers(filter, new QueryPagination({ size, searchAfter }), options);
+
+        allTransfers.push(...batch);
+      }
+
+      return allTransfers.filter((transfer) => transfer.canBeIgnored !== true);
+    } catch (error) {
+      this.logger.error(`Error fetching transfers for timestamp '${timestampMs}'`);
+      this.logger.error(error);
+      return [];
+    }
+  }
+
+  private async fetchEvents(timestampMs: number): Promise<Events[]> {
+    try {
+      const size = CustomSubscriptionsDataFetcher.batchSize;
+      const filter = new EventsFilter({ before: timestampMs, after: timestampMs });
+
+      const allEvents: Events[] = [];
+
+      let batch = await this.eventsService.getEvents(new QueryPagination({ size }), filter);
+      allEvents.push(...batch);
+
+      while (batch.length === size) {
+        const searchAfter = batch[batch.length - 1].searchAfter;
+        if (searchAfter == null) {
+          break;
+        }
+
+        batch = await this.eventsService.getEvents(new QueryPagination({ size, searchAfter }), filter);
+
+        allEvents.push(...batch);
+      }
+
+      return allEvents;
+    } catch (error) {
+      this.logger.error(`Error fetching events for timestamp '${timestampMs}'`);
+      this.logger.error(error);
+      return [];
+    }
+  }
+
+  // Transactions are the 'normal' subset of the operations returned for transfers. They are
+  // cloned so that clearing the type does not alter the objects broadcast on the transfers channel.
+  private extractTransactions(transfers: Transaction[]): Transaction[] {
+    const transactions: Transaction[] = [];
+
+    for (const transfer of transfers) {
+      if (transfer.type !== TransactionType.Transaction) {
+        continue;
+      }
+
+      const transaction = Object.assign(new TransactionDetailed(), transfer);
+      transaction.type = undefined;
+
+      transactions.push(transaction);
+    }
+
+    return transactions;
+  }
+}

src/crons/websocket/events.custom.gateway.ts

@@ -2,13 +2,10 @@ import { WebSocketGateway, WebSocketServer, SubscribeMessage, ConnectedSocket, M
 import { Server, Socket } from 'socket.io';
 import { UseFilters, UseInterceptors } from '@nestjs/common';
 import { OriginLogger } from '@multiversx/sdk-nestjs-common';
-import { QueryPagination } from 'src/common/entities/query.pagination';
 import { WsValidationPipe } from 'src/utils/ws-validation.pipe';
 import { WebsocketExceptionsFilter } from 'src/utils/ws-exceptions.filter';
 import { RoomKeyGenerator } from './room.key.generator';
-import { EventsService } from 'src/endpoints/events/events.service';
 import { EventsCustomSubscribePayload } from 'src/endpoints/events/entities/events.custom.subscribe';
-import { EventsFilter } from 'src/endpoints/events/entities/events.filter';
 import { Events } from 'src/endpoints/events/entities/events';
 import { LockingGuardInterceptor } from 'src/utils/locking.guard.interceptor';
 
@@ -22,10 +19,6 @@ export class EventsCustomGateway {
   @WebSocketServer()
   server!: Server;
 
-  constructor(
-    private readonly eventsService: EventsService,
-  ) { }
-
   @UseInterceptors(LockingGuardInterceptor)
   @SubscribeMessage('subscribeCustomEvents')
   async handleCustomSubscription(
@@ -57,29 +50,11 @@ export class EventsCustomGateway {
     return { status: 'unsubscribed' };
   }
 
-  async pushEventsForTimestampMs(timestampMs: number): Promise<void> {
+  pushEventsForTimestampMs(timestampMs: number, events: Events[]): void {
     try {
-      const allEvents: Events[] = [];
-      const size = 10000;
-      const filter = new EventsFilter({ before: timestampMs, after: timestampMs });
-
-      let batch = await this.eventsService.getEvents(new QueryPagination({ size }), filter);
-      allEvents.push(...batch);
-
-      while (batch.length === size) {
-        const searchAfter = batch[batch.length - 1].searchAfter;
-        if (searchAfter == null) {
-          break;
-        }
-
-        batch = await this.eventsService.getEvents(new QueryPagination({ size, searchAfter }), filter);
-
-        allEvents.push(...batch);
-      }
-
       const eventsFilteredForBroadcast: Map<string, Events[]> = new Map();
 
-      for (const event of allEvents) {
+      for (const event of events) {
         const roomKeys = RoomKeyGenerator.generate(
           EventsCustomGateway.keyPrefix,
           event,

src/crons/websocket/events.gateway.ts

@@ -28,7 +28,7 @@ export class EventsGateway {
         @ConnectedSocket() client: Socket,
         @MessageBody(new WsValidationPipe()) payload: EventsSubscribePayload,
     ) {
-        const filterIdentifier = JSON.stringify(payload);
+        const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
         const roomName = `${EventsGateway.keyPrefix}${filterIdentifier}`;
         if (!client.rooms.has(roomName)) {
             await client.join(roomName);

src/crons/websocket/network.gateway.ts

@@ -38,7 +38,7 @@ export class NetworkGateway {
   async pushStats() {
     if (this.server.sockets.adapter.rooms.has('statsRoom')) {
       try {
-        const stats = await this.networkService.getStats();
+        const stats = await this.networkService.getStats(true);
         this.server.to('statsRoom').emit('statsUpdate', stats);
       } catch (error) {
         this.logger.error(error);

src/crons/websocket/room.key.generator.ts

@@ -88,6 +88,17 @@ export class RoomKeyGenerator {
     return rooms;
   }
 
+  // Renaming a field can move it to a different position in the sorted key, so the key is
+  // rebuilt from its parsed form instead of being patched in place.
+  public static substitute(prefix: string, roomKey: string, from: string, to: string): string {
+    const { [from]: value, ...rest } = JSON.parse(roomKey.slice(prefix.length));
+    if (value === undefined) {
+      return roomKey;
+    }
+
+    return `${prefix}${this.deterministicStringify({ ...rest, [to]: value })}`;
+  }
+
   static deterministicStringify(obj: Record<string, any>): string {
     return JSON.stringify(
       Object.keys(obj)

src/crons/websocket/transaction.custom.gateway.ts

@@ -1,8 +1,5 @@
 import { WebSocketGateway, WebSocketServer, SubscribeMessage, ConnectedSocket, MessageBody } from '@nestjs/websockets';
 import { Server, Socket } from 'socket.io';
-import { TransactionService } from '../../endpoints/transactions/transaction.service';
-import { TransactionFilter } from '../../endpoints/transactions/entities/transaction.filter';
-import { QueryPagination } from 'src/common/entities/query.pagination';
 import { WsValidationPipe } from 'src/utils/ws-validation.pipe';
 import { WebsocketExceptionsFilter } from 'src/utils/ws-exceptions.filter';
 import { UseFilters, UseInterceptors } from '@nestjs/common';
@@ -20,10 +17,6 @@ export class TransactionsCustomGateway {
   @WebSocketServer()
   server!: Server;
 
-  constructor(
-    private readonly transactionService: TransactionService,
-  ) { }
-
   @UseInterceptors(LockingGuardInterceptor)
   @SubscribeMessage('subscribeCustomTransactions')
   async handleCustomSubscription(
@@ -52,28 +45,10 @@ export class TransactionsCustomGateway {
     return { status: 'unsubscribed' };
   }
 
-  async pushTransactionsForTimestampMs(timestampMs: number): Promise<void> {
+  pushTransactionsForTimestampMs(timestampMs: number, transactions: Transaction[]): void {
     try {
-      const allTransactions: Transaction[] = [];
-      const size = 10000;
-      const filter = new TransactionFilter({ before: timestampMs, after: timestampMs });
-
-      let batch = await this.transactionService.getTransactions(filter, new QueryPagination({ size }));
-      allTransactions.push(...batch);
-
-      while (batch.length === size) {
-        const searchAfter = batch[batch.length - 1].searchAfter;
-        if (searchAfter == null) {
-          break;
-        }
-
-        batch = await this.transactionService.getTransactions(filter, new QueryPagination({ size, searchAfter }));
-
-        allTransactions.push(...batch);
-      }
-
       const txFilteredForBroadcast: Map<string, Transaction[]> = new Map();
-      for (const transaction of allTransactions) {
+      for (const transaction of transactions) {
         const roomKeys = RoomKeyGenerator.generate(
           TransactionsCustomGateway.keyPrefix,
           transaction,

src/crons/websocket/transaction.gateway.ts

@@ -47,7 +47,7 @@ export class TransactionsGateway {
 
     TransactionFilter.validate(transactionFilter, payload.size || 25);
 
-    const filterIdentifier = JSON.stringify(payload);
+    const filterIdentifier = RoomKeyGenerator.deterministicStringify(payload);
     const roomName = `${TransactionsGateway.keyPrefix}${filterIdentifier}`;
 
     if (!client.rooms.has(roomName)) {