Pool count from gateway

src/common/gateway/gateway.service.ts

@@ -182,8 +182,22 @@ export class GatewayService {
     return new NftData(result.tokenData);
   }
 
-  async getTransactionPool(): Promise<TxPoolGatewayResponse> {
-    return await this.get(`transaction/pool?fields=*`, GatewayComponentRequest.transactionPool);
+  async getTransactionPool(): Promise<TxPoolGatewayResponse | null> {
+    try {
+      return await this.get(`transaction/pool?fields=*`, GatewayComponentRequest.transactionPool);
+    } catch (error: any) {
+      if (error?.message?.startsWith('maxContentLength size of')) {
+        return null;
+      }
+
+      throw error;
+    }
+  }
+
+  async getTransactionPoolCount(): Promise<number> {
+    const result = await this.get('transaction/pool/count', GatewayComponentRequest.transactionPool);
+
+    return Object.values<number>(result.txPoolCounts).reduce((total, count) => total + count, 0);
   }
 
   async getTransaction(txHash: string): Promise<Transaction | undefined> {

src/crons/cache.warmer/cache.warmer.service.ts

@@ -151,9 +151,19 @@ export class CacheWarmerService {
 
   @Lock({ name: 'Transaction pool invalidation', verbose: true })
   async handleTxPoolInvalidations() {
+    const ttl = this.apiConfigService.getTransactionPoolCacheWarmerTtlInSeconds();
     const pool = await this.poolService.getTxPoolRaw();
 
-    await this.invalidateKey(CacheInfo.TransactionPool.key, pool, this.apiConfigService.getTransactionPoolCacheWarmerTtlInSeconds());
+    const invalidations = [
+      this.invalidateKey(CacheInfo.TransactionPool.key, pool, ttl),
+    ];
+
+    if (pool == null) {
+      const count = await this.gatewayService.getTransactionPoolCount();
+      invalidations.push(this.invalidateKey(CacheInfo.TransactionPoolCount.key, count, ttl));
+    }
+
+    await Promise.all(invalidations);
   }
 
   @Cron('*/2 * * * *')

src/crons/websocket/pool.gateway.ts

@@ -10,6 +10,7 @@ import { QueryPagination } from 'src/common/entities/query.pagination';
 import { PoolSubscribePayload } from '../../endpoints/pool/entities/pool.subscribe';
 import { RoomKeyGenerator } from './room.key.generator';
 import { LockingGuardInterceptor } from 'src/utils/locking.guard.interceptor';
+import { PoolUpdateStatus } from '../../endpoints/pool/entities/pool.update.status';
 
 @UseFilters(WebsocketExceptionsFilter)
 @WebSocketGateway({ cors: { origin: '*' }, path: '/ws/subscription' })
@@ -63,20 +64,31 @@ export class PoolGateway {
                 type: filter.type,
             });
 
+            let status = PoolUpdateStatus.success;
+
             const [pool, poolCount] = await Promise.all([
                 this.poolService.getPool(
                     new QueryPagination({
                         from: filter.from,
                         size: filter.size,
                     }),
                     poolFilter,
-                ),
+                ).catch((error) => {
+                    this.logger.error(error);
+                    status = PoolUpdateStatus.internalServerError;
+                    return null;
+                }),
                 this.poolService.getPoolCount(poolFilter),
             ]);
 
-            this.server.to(roomName).emit("poolUpdate", { pool, poolCount });
+            if (pool == null && status === PoolUpdateStatus.success) {
+                status = PoolUpdateStatus.tooLarge;
+            }
+
+            this.server.to(roomName).emit("poolUpdate", { status, pool, poolCount });
         } catch (error) {
             this.logger.error(error);
+            this.server.to(roomName).emit("poolUpdate", { status: PoolUpdateStatus.internalServerError, pool: null, poolCount: null });
         }
     }
 

src/endpoints/pool/entities/pool.update.status.ts

@@ -0,0 +1,5 @@
+export enum PoolUpdateStatus {
+  success = 'success',
+  tooLarge = 'tooLarge',
+  internalServerError = 'internalServerError',
+}

src/endpoints/pool/entities/transaction.pool.too.large.exception.ts

@@ -0,0 +1,13 @@
+import { HttpException, HttpStatus } from "@nestjs/common";
+
+export class TransactionPoolTooLargeException extends HttpException {
+  static readonly code = 'transaction_pool_too_large';
+
+  constructor() {
+    super({
+      statusCode: HttpStatus.SERVICE_UNAVAILABLE,
+      code: TransactionPoolTooLargeException.code,
+      message: 'The transaction pool is too large to be displayed',
+    }, HttpStatus.SERVICE_UNAVAILABLE);
+  }
+}

src/endpoints/pool/pool.controller.ts

@@ -1,12 +1,13 @@
 import { ParseAddressAndMetachainPipe, ParseAddressPipe, ParseEnumPipe, ParseIntPipe, ParseTransactionHashPipe, ParseArrayPipe } from "@multiversx/sdk-nestjs-common";
 import { Controller, DefaultValuePipe, Get, NotFoundException, Param, Query } from "@nestjs/common";
-import { ApiExcludeEndpoint, ApiNotFoundResponse, ApiOkResponse, ApiOperation, ApiQuery, ApiTags } from "@nestjs/swagger";
+import { ApiExcludeEndpoint, ApiNotFoundResponse, ApiOkResponse, ApiOperation, ApiQuery, ApiServiceUnavailableResponse, ApiTags } from "@nestjs/swagger";
 import { PoolService } from "./pool.service";
 import { QueryPagination } from "src/common/entities/query.pagination";
 import { TransactionInPool } from "./entities/transaction.in.pool.dto";
 import { TransactionType } from "../transactions/entities/transaction.type";
 import { PoolFilter } from "./entities/pool.filter";
 import { ParseArrayPipeOptions } from "@multiversx/sdk-nestjs-common/lib/pipes/entities/parse.array.options";
+import { TransactionPoolTooLargeException } from "./entities/transaction.pool.too.large.exception";
 
 @Controller()
 @ApiTags('pool')
@@ -18,6 +19,7 @@ export class PoolController {
   @Get("/pool")
   @ApiOperation({ summary: 'Transactions pool', description: 'Returns the transactions that are currently in the memory pool.' })
   @ApiOkResponse({ type: TransactionInPool, isArray: true })
+  @ApiServiceUnavailableResponse({ description: 'The transaction pool is too large to be displayed' })
   @ApiQuery({ name: 'from', description: 'Number of items to skip for the result set', required: false })
   @ApiQuery({ name: 'size', description: 'Number of items to retrieve', required: false })
   @ApiQuery({ name: 'sender', description: 'Search in transaction pool by a specific sender', required: false })
@@ -37,14 +39,20 @@ export class PoolController {
     @Query('type', new ParseEnumPipe(TransactionType)) type?: TransactionType,
     @Query('function', new ParseArrayPipe(new ParseArrayPipeOptions({ allowEmptyString: true }))) functions?: string[],
   ): Promise<TransactionInPool[]> {
-    return await this.poolService.getPool(new QueryPagination({ from, size }), new PoolFilter({
+    const pool = await this.poolService.getPool(new QueryPagination({ from, size }), new PoolFilter({
       sender: sender,
       receiver: receiver,
       senderShard: senderShard,
       receiverShard: receiverShard,
       type: type,
       functions: functions,
     }));
+
+    if (pool == null) {
+      throw new TransactionPoolTooLargeException();
+    }
+
+    return pool;
   }
 
   @Get("/pool/count")
@@ -85,10 +93,15 @@ export class PoolController {
   @ApiOperation({ summary: 'Transaction from pool', description: 'Returns a transaction from the memory pool.' })
   @ApiOkResponse({ type: TransactionInPool })
   @ApiNotFoundResponse({ description: 'Transaction not found' })
+  @ApiServiceUnavailableResponse({ description: 'The transaction pool is too large to be displayed' })
   async getTransactionFromPool(
     @Param('txhash', ParseTransactionHashPipe) txHash: string,
   ): Promise<TransactionInPool> {
     const transaction = await this.poolService.getTransactionFromPool(txHash);
+    if (transaction === null) {
+      throw new TransactionPoolTooLargeException();
+    }
+
     if (transaction === undefined) {
       throw new NotFoundException('Transaction not found');
     }

src/endpoints/pool/pool.service.ts

@@ -12,6 +12,7 @@ import { TxInPoolFields } from "src/common/gateway/entities/tx.in.pool.fields";
 import { TransactionActionService } from "../transactions/transaction-action/transaction.action.service";
 import { Transaction } from "../transactions/entities/transaction";
 import { ApiUtils } from "@multiversx/sdk-nestjs-http";
+import { TransactionPoolTooLargeException } from "./entities/transaction.pool.too.large.exception";
 
 @Injectable()
 export class PoolService {
@@ -22,43 +23,72 @@ export class PoolService {
     private readonly transactionActionService: TransactionActionService,
   ) { }
 
-  async getTransactionFromPool(txHash: string): Promise<TransactionInPool | undefined> {
-    const pool = await this.getPoolWithFilters();
+  async getTransactionFromPool(txHash: string): Promise<TransactionInPool | undefined | null> {
+    const pool = await this.getTxPool();
+    if (pool == null) {
+      return null;
+    }
+
     return pool.find(tx => tx.txHash === txHash);
   }
 
   async getPoolCount(filter: PoolFilter): Promise<number> {
-    const pool = await this.getPoolWithFilters(filter);
-    return pool.length;
+    const pool = await this.getTxPool().catch(() => null);
+    if (pool != null) {
+      return this.applyFilters(pool, filter).length;
+    }
+
+    return await this.cacheService.getOrSet(
+      CacheInfo.TransactionPoolCount.key,
+      async () => await this.gatewayService.getTransactionPoolCount(),
+      CacheInfo.TransactionPoolCount.ttl,
+    );
   }
 
   async getPool(
     queryPagination: QueryPagination,
     filter?: PoolFilter,
-  ): Promise<TransactionInPool[]> {
+  ): Promise<TransactionInPool[] | null> {
     if (!this.apiConfigService.isTransactionPoolEnabled()) {
       return [];
     }
 
     const { from, size } = queryPagination;
-    const pool = await this.getPoolWithFilters(filter);
-    return pool.slice(from, from + size);
+    const pool = await this.getTxPool();
+    if (pool == null) {
+      return null;
+    }
+
+    return this.applyFilters(pool, filter).slice(from, from + size);
   }
 
   async getPoolWithFilters(
     filter?: PoolFilter,
   ): Promise<TransactionInPool[]> {
-    const pool = await this.cacheService.getOrSet(
+    const pool = await this.getTxPool();
+    if (pool == null) {
+      throw new TransactionPoolTooLargeException();
+    }
+
+    return this.applyFilters(pool, filter);
+  }
+
+  private async getTxPool(): Promise<TransactionInPool[] | null> {
+    return await this.cacheService.getOrSet(
       CacheInfo.TransactionPool.key,
       async () => await this.getTxPoolRaw(),
       CacheInfo.TransactionPool.ttl,
+      CacheInfo.TransactionPool.ttl,
+      true,
     );
-
-    return this.applyFilters(pool, filter);
   }
 
-  async getTxPoolRaw(): Promise<TransactionInPool[]> {
+  async getTxPoolRaw(): Promise<TransactionInPool[] | null> {
     const pool = await this.gatewayService.getTransactionPool();
+    if (pool == null) {
+      return null;
+    }
+
     return this.parseTransactions(pool);
   }
 

src/test/unit/services/cache.warmer.pool.spec.ts

@@ -0,0 +1,51 @@
+import { Locker, LockResult } from "@multiversx/sdk-nestjs-common";
+import { CacheWarmerService } from "src/crons/cache.warmer/cache.warmer.service";
+import { TransactionType } from "src/endpoints/transactions/entities/transaction.type";
+import { CacheInfo } from "src/utils/cache.info";
+
+describe('CacheWarmerService transaction pool', () => {
+  const ttl = 10;
+  const pool = [{ txHash: 'a', type: TransactionType.Transaction }];
+
+  let warmer: CacheWarmerService;
+  let poolService: any;
+  let gatewayService: any;
+  let cachingService: any;
+
+  beforeEach(() => {
+    jest.spyOn(Locker, 'lock').mockImplementation(async (_key: string, func: () => Promise<void>) => {
+      await func();
+      return LockResult.SUCCESS;
+    });
+
+    poolService = { getTxPoolRaw: jest.fn().mockResolvedValue(pool) };
+    gatewayService = { getTransactionPoolCount: jest.fn().mockResolvedValue(12) };
+    cachingService = { set: jest.fn() };
+
+    warmer = Object.assign(Object.create(CacheWarmerService.prototype), {
+      poolService,
+      gatewayService,
+      cachingService,
+      apiConfigService: { getTransactionPoolCacheWarmerTtlInSeconds: () => ttl },
+      clientProxy: { emit: jest.fn() },
+    });
+  });
+
+  it('should warm only the pool when it can be read', async () => {
+    await warmer.handleTxPoolInvalidations();
+
+    expect(cachingService.set).toHaveBeenCalledTimes(1);
+    expect(cachingService.set).toHaveBeenCalledWith(CacheInfo.TransactionPool.key, pool, ttl);
+    expect(gatewayService.getTransactionPoolCount).not.toHaveBeenCalled();
+  });
+
+  it('should warm the pool as null and refresh the count from the gateway when the pool is too large', async () => {
+    poolService.getTxPoolRaw.mockResolvedValue(null);
+
+    await warmer.handleTxPoolInvalidations();
+
+    expect(cachingService.set).toHaveBeenCalledTimes(2);
+    expect(cachingService.set).toHaveBeenCalledWith(CacheInfo.TransactionPool.key, null, ttl);
+    expect(cachingService.set).toHaveBeenCalledWith(CacheInfo.TransactionPoolCount.key, 12, ttl);
+  });
+});

src/test/unit/services/gateway.pool.count.spec.ts

@@ -0,0 +1,35 @@
+import { GatewayService } from "src/common/gateway/gateway.service";
+
+describe('GatewayService transaction pool', () => {
+  let gatewayService: GatewayService;
+  let apiService: any;
+
+  beforeEach(() => {
+    apiService = {
+      get: jest.fn().mockResolvedValue({ data: { data: { txPoolCounts: { '0': 1, '1': 2, '2': 3, '4294967295': 4 } } } }),
+    };
+
+    gatewayService = new GatewayService(
+      { getGatewayUrl: () => 'https://gateway', getSnapshotlessGatewayUrl: () => undefined } as any,
+      apiService,
+    );
+    Object.assign(gatewayService, { eventEmitter: { emit: jest.fn() } });
+  });
+
+  it('should read a pool over the response size limit as null', async () => {
+    apiService.get.mockRejectedValue({ message: 'maxContentLength size of 2097152 exceeded' });
+
+    expect(await gatewayService.getTransactionPool()).toBeNull();
+  });
+
+  it('should keep throwing other failures of the pool', async () => {
+    apiService.get.mockRejectedValue({ message: 'connect ECONNREFUSED' });
+
+    await expect(gatewayService.getTransactionPool()).rejects.toEqual({ message: 'connect ECONNREFUSED' });
+  });
+
+  it('should sum the counts of every shard', async () => {
+    expect(await gatewayService.getTransactionPoolCount()).toStrictEqual(10);
+    expect(apiService.get).toHaveBeenCalledWith('https://gateway/transaction/pool/count', expect.anything(), undefined);
+  });
+});