Monitor chain for capacity stake/unstake (#329)

Add Capacity event monitoring to the blockchain scanner, so that we can
pause/unpause the job queue as Capacity becomes available or
unavailable, based on staking. Previously, if the queue was paused due
to low Capacity, we would have to wait for the next epoch, even if more
Capacity was staked.

- [x] account-service
- [x] content-publishing-service
- ~[ ] content-watcher-service~ (not needed)
- [x] graph-service

Closes: #296 
Closes: #267

---------

Co-authored-by: Matthew Orris <[email protected]>

Author: JoeCap08055

.tool-versions

@@ -1,2 +1,2 @@
-nodejs 20.12.2
+nodejs 20.16.0
 make 4.3

package-lock.json

@@ -1,6 +1,5 @@
-import { Controller, Get, Post, HttpCode, HttpStatus, Logger, Param, HttpException, Body, Put } from '@nestjs/common';
+import { Controller, Get, Post, HttpCode, HttpStatus, Logger, Param, HttpException, Body } from '@nestjs/common';
 import { ApiBody, ApiOkResponse, ApiOperation, ApiTags } from '@nestjs/swagger';
-import type { HandleResponse } from '@frequency-chain/api-augment/interfaces';
 import { TransactionType } from '#lib/types/enums';
 import { HandlesService } from '#api/services/handles.service';
 import { EnqueueService } from '#lib/services/enqueue-request.service';

services/account/apps/api/src/controllers/v1/handles-v1.controller.ts

@@ -5,6 +5,12 @@ import { EventEmitter2 } from '@nestjs/event-emitter';
 import { redisReady } from '#lib/utils/redis';
 import { WorkerModule } from './worker.module';
 
+// Monkey-patch BigInt so that JSON.stringify will work
+// eslint-disable-next-line
+BigInt.prototype['toJSON'] = function () {
+  return this.toString();
+};
+
 const logger = new Logger('worker_main');
 
 // bootstrap() does not have a main loop to keep the process alive.

services/account/apps/worker/src/main.ts

@@ -1,11 +1,10 @@
 import { Module } from '@nestjs/common';
 import { BlockchainModule } from '#lib/blockchain/blockchain.module';
-import { EnqueueService } from '#lib/services/enqueue-request.service';
 import { TxnNotifierService } from './notifier.service';
 
 @Module({
   imports: [BlockchainModule],
-  providers: [EnqueueService, TxnNotifierService],
-  exports: [EnqueueService, TxnNotifierService],
+  providers: [TxnNotifierService],
+  exports: [],
 })
 export class TxnNotifierModule {}

services/account/apps/worker/src/transaction_notifier/notifier.module.ts

@@ -8,14 +8,15 @@ import { BlockchainService } from '#lib/blockchain/blockchain.service';
 import { TransactionType } from '#lib/types/enums';
 import { SECONDS_PER_BLOCK, TxWebhookRsp, RedisUtils } from 'libs/common/src';
 import { createWebhookRsp } from '#worker/transaction_notifier/notifier.service.helper.createWebhookRsp';
-import { BlockchainScannerService, NullScanError } from '#lib/utils/blockchain-scanner.service';
+import { BlockchainScannerService } from '#lib/utils/blockchain-scanner.service';
 import { SchedulerRegistry } from '@nestjs/schedule';
-import { BlockHash } from '@polkadot/types/interfaces';
+import { SignedBlock } from '@polkadot/types/interfaces';
 import { HexString } from '@polkadot/util/types';
 import { ITxStatus } from '#lib/interfaces/tx-status.interface';
 import { FrameSystemEventRecord } from '@polkadot/types/lookup';
 import { ConfigService } from '#lib/config/config.service';
 import { QueueConstants } from '#lib/queues';
+import { CapacityCheckerService } from '#lib/blockchain/capacity-checker.service';
 
 @Injectable()
 export class TxnNotifierService
@@ -47,9 +48,13 @@ export class TxnNotifierService
     private readonly schedulerRegistry: SchedulerRegistry,
     @InjectRedis() cacheManager: Redis,
     private readonly configService: ConfigService,
+    private readonly capacityService: CapacityCheckerService,
   ) {
     super(cacheManager, blockchainService, new Logger(TxnNotifierService.prototype.constructor.name));
     this.scanParameters = { onlyFinalized: this.configService.trustUnfinalizedBlocks };
+    this.registerChainEventHandler(['capacity.UnStaked', 'capacity.Staked'], () =>
+      this.capacityService.checkForSufficientCapacity(),
+    );
   }
 
   public get intervalName() {
@@ -72,25 +77,6 @@ export class TxnNotifierService
     }
   }
 
-  protected async checkInitialScanParameters(): Promise<void> {
-    const pendingTxns = await this.cacheManager.hlen(RedisUtils.TXN_WATCH_LIST_KEY);
-    if (pendingTxns === 0) {
-      throw new NullScanError('No pending extrinsics; no scan will be performed');
-    }
-
-    return super.checkInitialScanParameters();
-  }
-
-  protected async checkScanParameters(blockNumber: number, blockHash: BlockHash): Promise<void> {
-    const pendingTxns = await this.cacheManager.hlen(RedisUtils.TXN_WATCH_LIST_KEY);
-
-    if (pendingTxns === 0) {
-      throw new NullScanError('No pending extrinsics; terminating current scan iteration');
-    }
-
-    return super.checkScanParameters(blockNumber, blockHash);
-  }
-
   public async getLastSeenBlockNumber(): Promise<number> {
     let blockNumber = await super.getLastSeenBlockNumber();
     const pendingTxns = await this.cacheManager.hvals(RedisUtils.TXN_WATCH_LIST_KEY);
@@ -107,15 +93,16 @@ export class TxnNotifierService
     return blockNumber;
   }
 
-  async processCurrentBlock(currentBlockHash: BlockHash, currentBlockNumber: number): Promise<void> {
+  async processCurrentBlock(currentBlock: SignedBlock, blockEvents: FrameSystemEventRecord[]): Promise<void> {
+    const currentBlockNumber = currentBlock.block.header.number.toNumber();
+
     // Get set of tx hashes to monitor from cache
     const pendingTxns = (await this.cacheManager.hvals(RedisUtils.TXN_WATCH_LIST_KEY)).map(
       (val) => JSON.parse(val) as ITxStatus,
     );
 
-    const block = await this.blockchainService.getBlock(currentBlockHash);
     const extrinsicIndices: [HexString, number][] = [];
-    block.block.extrinsics.forEach((extrinsic, index) => {
+    currentBlock.block.extrinsics.forEach((extrinsic, index) => {
       if (pendingTxns.some(({ txHash }) => txHash === extrinsic.hash.toHex())) {
         extrinsicIndices.push([extrinsic.hash.toHex(), index]);
       }
@@ -124,9 +111,9 @@ export class TxnNotifierService
     let pipeline = this.cacheManager.multi({ pipeline: true });
 
     if (extrinsicIndices.length > 0) {
-      const at = await this.blockchainService.api.at(currentBlockHash);
+      const at = await this.blockchainService.api.at(currentBlock.block.header.hash);
       const epoch = (await at.query.capacity.currentEpoch()).toNumber();
-      const events: FrameSystemEventRecord[] = (await at.query.system.events()).filter(
+      const events: FrameSystemEventRecord[] = blockEvents.filter(
         ({ phase }) => phase.isApplyExtrinsic && extrinsicIndices.some((index) => phase.asApplyExtrinsic.eq(index)),
       );
 
@@ -154,7 +141,7 @@ export class TxnNotifierService
           const moduleError = dispatchError.registry.findMetaError(moduleThatErrored);
           this.logger.error(`Extrinsic failed with error: ${JSON.stringify(moduleError)}`);
         } else if (successEvent) {
-          this.logger.verbose(`Successfully found transaction ${txHash} in block ${currentBlockHash}`);
+          this.logger.verbose(`Successfully found transaction ${txHash} in block ${currentBlockNumber}`);
           const webhook = await this.getWebhook();
           let webhookResponse: Partial<TxWebhookRsp> = {};
           webhookResponse.referenceId = txStatus.referenceId;

services/account/apps/worker/src/transaction_notifier/notifier.service.ts

@@ -8,7 +8,7 @@ import { SubmittableExtrinsic } from '@polkadot/api-base/types';
 import { Codec, ISubmittableResult } from '@polkadot/types/types';
 import { MILLISECONDS_PER_SECOND } from 'time-constants';
 import { SchedulerRegistry } from '@nestjs/schedule';
-import { BlockchainService } from '#lib/blockchain/blockchain.service';
+import { BlockchainService, ICapacityInfo } from '#lib/blockchain/blockchain.service';
 import { createKeys } from '#lib/blockchain/create-keys';
 import { NonceService } from '#lib/services/nonce.service';
 import { TransactionType } from '#lib/types/enums';
@@ -18,6 +18,12 @@ import { RedisUtils, TransactionData } from 'libs/common/src';
 import { ConfigService } from '#lib/config/config.service';
 import { ITxStatus } from '#lib/interfaces/tx-status.interface';
 import { HexString } from '@polkadot/util/types';
+import {
+  CAPACITY_AVAILABLE_EVENT,
+  CAPACITY_EXHAUSTED_EVENT,
+  CapacityCheckerService,
+} from '#lib/blockchain/capacity-checker.service';
+import { OnEvent } from '@nestjs/event-emitter';
 
 export const SECONDS_PER_BLOCK = 12;
 const CAPACITY_EPOCH_TIMEOUT_NAME = 'capacity_check';
@@ -29,7 +35,7 @@ const CAPACITY_EPOCH_TIMEOUT_NAME = 'capacity_check';
 @Processor(QueueConstants.TRANSACTION_PUBLISH_QUEUE)
 export class TransactionPublisherService extends BaseConsumer implements OnApplicationShutdown {
   public async onApplicationBootstrap() {
-    await this.checkCapacity();
+    await this.capacityCheckerService.checkForSufficientCapacity();
   }
 
   public async onApplicationShutdown(_signal?: string | undefined): Promise<void> {
@@ -48,6 +54,7 @@ export class TransactionPublisherService extends BaseConsumer implements OnAppli
     private blockchainService: BlockchainService,
     private nonceService: NonceService,
     private schedulerRegistry: SchedulerRegistry,
+    private capacityCheckerService: CapacityCheckerService,
   ) {
     super();
   }
@@ -61,7 +68,7 @@ export class TransactionPublisherService extends BaseConsumer implements OnAppli
     let txHash: HexString;
     try {
       // Check capacity first; if out of capacity, send job back to queue
-      if (!(await this.checkCapacity())) {
+      if (!(await this.capacityCheckerService.checkForSufficientCapacity())) {
         job.moveToDelayed(Date.now(), job.token); // fake delay, we just want to avoid processing the current job if we're out of capacity
         throw new DelayedError();
       }
@@ -114,11 +121,10 @@ export class TransactionPublisherService extends BaseConsumer implements OnAppli
       obj[txHash] = JSON.stringify(status);
       this.cacheManager.hset(RedisUtils.TXN_WATCH_LIST_KEY, obj);
     } catch (error: any) {
-      if (error instanceof DelayedError) {
-        throw error;
+      if (!(error instanceof DelayedError)) {
+        this.logger.error('Unknown error encountered: ', error, error?.stack);
       }
 
-      this.logger.error('Unknown error encountered: ', error, error?.stack);
       throw error;
     }
   }
@@ -183,89 +189,59 @@ export class TransactionPublisherService extends BaseConsumer implements OnAppli
     }
   }
 
-  /**
-   * Checks the capacity of the account publisher and takes appropriate actions based on the capacity status.
-   * If the capacity is exhausted, it pauses the account change publish queue and sets a timeout to check the capacity again.
-   * If the capacity is refilled, it resumes the account change publish queue and clears the timeout.
-   * If any jobs failed due to low balance/no capacity, it retries them.
-   * If any error occurs during the capacity check, it logs the error.
-   */
-  private async checkCapacity(): Promise<boolean> {
-    let outOfCapacity = false;
+  @OnEvent(CAPACITY_EXHAUSTED_EVENT)
+  public async handleCapacityExhausted(capacityInfo: ICapacityInfo) {
+    await this.transactionPublishQueue.pause();
+    const blocksRemaining = capacityInfo.nextEpochStart - capacityInfo.currentBlockNumber;
+    const epochTimeout = blocksRemaining * SECONDS_PER_BLOCK * MILLISECONDS_PER_SECOND;
+    // Avoid spamming the log
+    if (!(await this.transactionPublishQueue.isPaused())) {
+      this.logger.warn(
+        `Capacity Exhausted: Pausing account change publish queue until next epoch: ${epochTimeout / 1000} seconds`,
+      );
+    }
     try {
-      const { capacityLimit } = this.configService;
-      const capacityInfo = await this.blockchainService.capacityInfo(this.configService.providerId);
-      const { remainingCapacity } = capacityInfo;
-      const { currentEpoch } = capacityInfo;
-      const epochCapacityKey = `epochCapacity:${currentEpoch}`;
-      const epochUsedCapacity = BigInt((await this.cacheManager.get(epochCapacityKey)) ?? 0); // Fetch capacity used by the service
-      outOfCapacity = remainingCapacity <= 0n;
-
-      if (!outOfCapacity) {
-        this.logger.debug(`      Capacity remaining: ${remainingCapacity}`);
-        if (capacityLimit.type === 'percentage') {
-          const capacityLimitPercentage = BigInt(capacityLimit.value);
-          const capacityLimitThreshold = (capacityInfo.totalCapacityIssued * capacityLimitPercentage) / 100n;
-          this.logger.debug(`Capacity limit threshold: ${capacityLimitThreshold}`);
-          if (epochUsedCapacity >= capacityLimitThreshold) {
-            outOfCapacity = true;
-            this.logger.warn(`Capacity threshold reached: used ${epochUsedCapacity} of ${capacityLimitThreshold}`);
-          }
-        } else if (epochUsedCapacity >= capacityLimit.value) {
-          outOfCapacity = true;
-          this.logger.warn(`Capacity threshold reached: used ${epochUsedCapacity} of ${capacityLimit.value}`);
-        }
+      // Check if a timeout with the same name already exists
+      if (this.schedulerRegistry.doesExist('timeout', CAPACITY_EPOCH_TIMEOUT_NAME)) {
+        // If it does, delete it
+        this.schedulerRegistry.deleteTimeout(CAPACITY_EPOCH_TIMEOUT_NAME);
       }
 
-      if (outOfCapacity) {
-        await this.transactionPublishQueue.pause();
-        const blocksRemaining = capacityInfo.nextEpochStart - capacityInfo.currentBlockNumber;
-        const epochTimeout = blocksRemaining * SECONDS_PER_BLOCK * MILLISECONDS_PER_SECOND;
-        this.logger.warn(
-          `Capacity Exhausted: Pausing account change publish queue until next epoch: ${epochTimeout / 1000} seconds`,
-        );
-        try {
-          // Check if a timeout with the same name already exists
-          if (this.schedulerRegistry.doesExist('timeout', CAPACITY_EPOCH_TIMEOUT_NAME)) {
-            // If it does, delete it
-            this.schedulerRegistry.deleteTimeout(CAPACITY_EPOCH_TIMEOUT_NAME);
-          }
-
-          // Add the new timeout
-          this.schedulerRegistry.addTimeout(
-            CAPACITY_EPOCH_TIMEOUT_NAME,
-            setTimeout(() => this.checkCapacity(), epochTimeout),
-          );
-        } catch (err) {
-          // Handle any errors
-          console.error(err);
-        }
-      } else {
-        this.logger.verbose('Capacity Available: Resuming account change publish queue and clearing timeout');
-        // Get the failed jobs and check if they failed due to capacity
-        const failedJobs = await this.transactionPublishQueue.getFailed();
-        const capacityFailedJobs = failedJobs.filter((job) =>
-          job.failedReason?.includes('1010: Invalid Transaction: Inability to pay some fees'),
-        );
-        // Retry the failed jobs
-        await Promise.all(
-          capacityFailedJobs.map(async (job) => {
-            this.logger.debug(`Retrying job ${job.id}`);
-            job.retry();
-          }),
-        );
-        try {
-          this.schedulerRegistry.deleteTimeout(CAPACITY_EPOCH_TIMEOUT_NAME);
-        } catch (err) {
-          // ignore
-        }
+      // Add the new timeout
+      this.schedulerRegistry.addTimeout(
+        CAPACITY_EPOCH_TIMEOUT_NAME,
+        setTimeout(() => this.capacityCheckerService.checkForSufficientCapacity(), epochTimeout),
+      );
+    } catch (err) {
+      // Handle any errors
+      console.error(err);
+    }
+  }
 
-        await this.transactionPublishQueue.resume();
-      }
+  @OnEvent(CAPACITY_AVAILABLE_EVENT)
+  public async handleCapacityAvailable() {
+    // Avoid spamming the log
+    if (await this.transactionPublishQueue.isPaused()) {
+      this.logger.verbose('Capacity Available: Resuming account change publish queue and clearing timeout');
+    }
+    // Get the failed jobs and check if they failed due to capacity
+    const failedJobs = await this.transactionPublishQueue.getFailed();
+    const capacityFailedJobs = failedJobs.filter((job) =>
+      job.failedReason?.includes('1010: Invalid Transaction: Inability to pay some fees'),
+    );
+    // Retry the failed jobs
+    await Promise.all(
+      capacityFailedJobs.map(async (job) => {
+        this.logger.debug(`Retrying job ${job.id}`);
+        job.retry();
+      }),
+    );
+    try {
+      this.schedulerRegistry.deleteTimeout(CAPACITY_EPOCH_TIMEOUT_NAME);
     } catch (err) {
-      this.logger.error('Caught error in checkCapacity', err);
+      // ignore
     }
 
-    return !outOfCapacity;
+    await this.transactionPublishQueue.resume();
   }
 }

services/account/apps/worker/src/transaction_publisher/publisher.service.ts

@@ -62,6 +62,6 @@ import { TransactionPublisherModule } from './transaction_publisher/publisher.mo
     TxnNotifierModule,
   ],
   providers: [ProviderWebhookService, NonceService],
-  exports: [],
+  exports: [EventEmitterModule],
 })
 export class WorkerModule {}

services/account/apps/worker/src/worker.module.ts

@@ -56,7 +56,7 @@ HEALTH_CHECK_SUCCESS_THRESHOLD=10
 # Maximum amount of provider capacity this app is allowed to use (per epoch)
 #     type: 'percentage' | 'amount'
 #     value: number (may be percentage, ie '80', or absolute amount of capacity)
-CAPACITY_LIMIT='{"type":"percentage", "value":80}'
+CAPACITY_LIMIT='{"serviceLimit":{"type":"percentage","value":"80"}}'
 
 # URL for the Sign-In With Frequency UI
 SIWF_URL=https://projectlibertylabs.github.io/siwf/ui

services/account/env.template

@@ -0,0 +1,6 @@
+/* eslint-disable func-names */
+/* eslint-disable no-extend-native */
+// eslint-disable-next-line dot-notation
+BigInt.prototype['toJSON'] = function () {
+  return this.toString();
+};