feat: Close PDAs for fills from our relayer (across-protocol#2381)

* feat: Close PDAs for fills from our relayer

* Log invalidFills without deposit

* Check if PDA is already closed and enable SVM addresses to be monitored

* lint fix

* Use one array of Addresses for both EVM and SVM Addresses for monitored relayers

* Fix MonitorClientHelper

* Remove unused interface

* Add simulate flag to closePDAs bot

* Lint fix

* Bump SDK

* Lint fix

* Exit early if monitoredAddress is not EVM

* Log invalidFills only if there are any

* Remove unnecessary if statement

* Remove unused invalidFills reporting for closing PDAs

* Remove unnecessary comment

* Lint fix

* reportRelayerBalances fix

Author: dijanin-brat

package.json

@@ -12,7 +12,7 @@
   "dependencies": {
     "@across-protocol/constants": "^3.1.68",
     "@across-protocol/contracts": "^4.1.0",
-    "@across-protocol/sdk": "4.3.40",
+    "@across-protocol/sdk": "4.3.44",
     "@arbitrum/sdk": "^4.0.2",
     "@aws-sdk/client-kms": "^3.592.0",
     "@aws-sdk/client-s3": "^3.592.0",
@@ -74,6 +74,7 @@
     "run-disputer": "DISPUTER_ENABLED=true node ./dist/index.js --dataworker",
     "run-executor": "EXECUTOR_ENABLED=true node ./dist/index.js --dataworker",
     "run-proposer": "PROPOSER_ENABLED=true node ./dist/index.js --dataworker",
+    "run-monitor": "node ./dist/index.js --monitor",
     "run-finalizer": "node ./dist/index.js --finalizer",
     "relay": "node ./dist/index.js --relayer",
     "deposit": "yarn ts-node ./scripts/spokepool.ts deposit",

src/clients/TokenTransferClient.ts

@@ -48,7 +48,8 @@ export class TokenTransferClient {
     const chainIds = Object.keys(this.providerByChainIds).map(Number);
     for (const chainId of chainIds) {
       const tokenContracts = tokenContractsByChainId[chainId];
-      for (const monitoredAddress of this.monitoredAddresses) {
+      const evmMonitoredAddresses = this.monitoredAddresses.filter((address) => address.isEVM());
+      for (const monitoredAddress of evmMonitoredAddresses) {
         const transferEventsList = await Promise.all(
           tokenContracts.map((tokenContract) =>
             this.querySendAndReceiveEvents(tokenContract, monitoredAddress, searchConfigByChainIds[chainId])

src/dataworker/Dataworker.ts

@@ -3185,7 +3185,8 @@ export class Dataworker {
     });
 
     // Get the slow fill information.
-    const relayDataHash = getRelayDataHash(leaf.relayData, leaf.chainId);
+    const messageHash = getMessageHash(leaf.relayData.message);
+    const relayDataHash = getRelayDataHash({ ...leaf.relayData, messageHash }, leaf.chainId);
 
     // Construct the slow fill instruction.
     const executeSlowFillIx = SvmSpokeClient.getExecuteSlowRelayLeafInstruction({

src/interfaces/index.ts

@@ -70,5 +70,6 @@ export type Refund = interfaces.Refund;
 export type RunningBalances = interfaces.RunningBalances;
 export type TokensBridged = interfaces.TokensBridged;
 export const { FillType, FillStatus } = interfaces;
+export type FillStatus = interfaces.FillStatus;
 
 export type CachingMechanismInterface = interfaces.CachingMechanismInterface;

src/libexec/RelayerSpokePoolListenerSVM.ts

@@ -83,7 +83,7 @@ async function scrapeEvents(
 ): Promise<void> {
   const provider = eventsClient.getRpc();
   const [{ timestamp: currentTime }, ...events] = await Promise.all([
-    arch.svm.getNearestSlotTime(provider, logger),
+    arch.svm.getNearestSlotTime(provider, { commitment: "confirmed" }, logger),
     ...eventNames.map((eventName) => _scrapeEvents(chain, eventsClient, eventName, { ...opts, to: opts.to }, logger)),
   ]);
 
@@ -173,7 +173,11 @@ async function run(argv: string[]): Promise<void> {
 
   const provider = getSvmProvider(await getRedisCache());
   const blockFinder = undefined;
-  const { slot: latestSlot, timestamp: now } = await arch.svm.getNearestSlotTime(provider, logger);
+  const { slot: latestSlot, timestamp: now } = await arch.svm.getNearestSlotTime(
+    provider,
+    { commitment: "confirmed" },
+    logger
+  );
 
   const deploymentBlock = getDeploymentBlockNumber("SvmSpoke", chainId);
   let startSlot = latestSlot;

src/monitor/Monitor.ts

@@ -5,6 +5,7 @@ import {
   BundleAction,
   DepositWithBlock,
   FillStatus,
+  FillWithBlock,
   L1Token,
   RelayerBalanceReport,
   RelayerBalanceTable,
@@ -55,12 +56,21 @@ import {
   getBinanceApiClient,
   getBinanceWithdrawalLimits,
   chainIsEvm,
+  getFillStatusPda,
+  getKitKeypairFromEvmSigner,
+  getRelayDataFromFill,
 } from "../utils";
 import { MonitorClients, updateMonitorClients } from "./MonitorClientHelper";
 import { MonitorConfig } from "./MonitorConfig";
 import { CombinedRefunds, getImpliedBundleBlockRanges } from "../dataworker/DataworkerUtils";
 import { PUBLIC_NETWORKS, TOKEN_EQUIVALENCE_REMAPPING } from "@across-protocol/constants";
 import { utils as sdkUtils, arch } from "@across-protocol/sdk";
+import {
+  address,
+  fetchEncodedAccount,
+  getBase64EncodedWireTransaction,
+  signTransactionMessageWithSigners,
+} from "@solana/kit";
 
 // 60 minutes, which is the length of the challenge window, so if a rebalance takes longer than this to finalize,
 // then its finalizing after the subsequent challenge period has started, which is sub-optimal.
@@ -481,6 +491,11 @@ export class Monitor {
     const l1Tokens = this.getL1TokensForRelayerBalancesReport();
     for (const relayer of this.monitorConfig.monitoredRelayers) {
       for (const chainId of this.monitorChains) {
+        // If the monitored relayer address is invalid on the monitored chain (e.g. the monitored relayer is a base58 address while the chain ID is mainnet),
+        // then there is no balance to update in this loop.
+        if (!relayer.isValidOn(chainId)) {
+          continue;
+        }
         const l2ToL1Tokens = this.getL2ToL1TokenMap(l1Tokens, chainId);
         const l2TokenAddresses = Object.keys(l2ToL1Tokens);
         const tokenBalances = await this._getBalances(
@@ -1183,6 +1198,90 @@ export class Monitor {
     }
   }
 
+  async closePDAs(): Promise<void> {
+    const simulate = process.env["SEND_TRANSACTIONS"] !== "true";
+    const svmSpokePoolClient = this.clients.spokePoolClients[CHAIN_IDs.SOLANA];
+    if (!isSVMSpokePoolClient(svmSpokePoolClient)) {
+      return;
+    }
+    const fills: FillWithBlock[] = [];
+    for (const relayers of this.monitorConfig.monitoredRelayers) {
+      if (relayers.isSVM()) {
+        const relayerFills = svmSpokePoolClient.getFillsForRelayer(relayers);
+        fills.push(...relayerFills);
+      }
+    }
+    const spokePoolProgramId = address(svmSpokePoolClient.spokePoolAddress.toBase58());
+    const signer = await getKitKeypairFromEvmSigner(this.clients.hubPoolClient.hubPool.signer);
+    const svmRpc = svmSpokePoolClient.svmEventsClient.getRpc();
+
+    for (const fill of fills) {
+      const relayData = getRelayDataFromFill(fill);
+      const relayDataWithMessageHash = {
+        ...relayData,
+        messageHash: fill.messageHash,
+      };
+      const fillStatus = await svmSpokePoolClient.relayFillStatus(relayDataWithMessageHash, fill.destinationChainId);
+      // If fill PDA should not be closed, skip.
+      if (!this._shouldCloseFillPDA(fillStatus, fill.fillDeadline, svmSpokePoolClient.getCurrentTime())) {
+        this.logger.info({
+          at: "Monitor#closePDAs",
+          message: `Not ready to close PDA for fill ${fill.txnRef}`,
+          fill,
+        });
+        continue;
+      }
+
+      const fillStatusPda = await getFillStatusPda(
+        spokePoolProgramId,
+        relayDataWithMessageHash,
+        fill.destinationChainId
+      );
+      // Check if PDA is already closed
+      const fillStatusPdaAccount = await fetchEncodedAccount(svmRpc, fillStatusPda);
+      if (!fillStatusPdaAccount.exists) {
+        continue;
+      }
+
+      const closePdaInstruction = await arch.svm.createCloseFillPdaInstruction(signer, svmRpc, fillStatusPda);
+      const signedTransaction = await signTransactionMessageWithSigners(closePdaInstruction);
+      const encodedTransaction = getBase64EncodedWireTransaction(signedTransaction);
+
+      if (simulate) {
+        const result = await svmRpc
+          .simulateTransaction(encodedTransaction, {
+            encoding: "base64",
+          })
+          .send();
+        if (result.value.err) {
+          this.logger.error({
+            at: "Monitor#closePDAs",
+            message: `Failed to close PDA for fill ${fill.txnRef}`,
+            error: result.value.err,
+          });
+        }
+        continue;
+      }
+
+      try {
+        await svmRpc
+          .sendTransaction(encodedTransaction, { preflightCommitment: "confirmed", encoding: "base64" })
+          .send();
+
+        this.logger.info({
+          at: "Monitor#closePDAs",
+          message: `Closed PDA ${fillStatusPda} for fill ${fill.txnRef}`,
+        });
+      } catch (err) {
+        this.logger.error({
+          at: "Monitor#closePDAs",
+          message: `Failed to close PDA for fill ${fill.txnRef}`,
+          error: err,
+        });
+      }
+    }
+  }
+
   async updateLatestAndFutureRelayerRefunds(relayerBalanceReport: RelayerBalanceReport): Promise<void> {
     const validatedBundleRefunds: CombinedRefunds[] =
       await this.clients.bundleDataClient.getPendingRefundsFromValidBundles();
@@ -1384,7 +1483,11 @@ export class Monitor {
         this.spokePoolsBlocks[chainId].endingBlock = endingBlock;
       } else if (isSVMSpokePoolClient(spokePoolClient)) {
         const svmProvider = await spokePoolClient.svmEventsClient.getRpc();
-        const { slot: latestSlot } = await arch.svm.getNearestSlotTime(svmProvider, spokePoolClient.logger);
+        const { slot: latestSlot } = await arch.svm.getNearestSlotTime(
+          svmProvider,
+          { commitment: "confirmed" },
+          spokePoolClient.logger
+        );
         const endingBlock = this.monitorConfig.spokePoolsBlocks[chainId]?.endingBlock;
         this.monitorConfig.spokePoolsBlocks[chainId] ??= { startingBlock: undefined, endingBlock: undefined };
         if (this.monitorConfig.pollingDelay === 0) {
@@ -1497,4 +1600,8 @@ export class Monitor {
     }
     return false;
   }
+
+  private _shouldCloseFillPDA(fillStatus: FillStatus, fillDeadline: number, currentTime: number): boolean {
+    return fillStatus === FillStatus.Filled && currentTime > fillDeadline;
+  }
 }

src/monitor/MonitorConfig.ts

@@ -21,6 +21,7 @@ export interface BotModes {
   unknownRootBundleCallersEnabled: boolean; // Monitors relay related events triggered by non-whitelisted addresses
   spokePoolBalanceReportEnabled: boolean;
   binanceWithdrawalLimitsEnabled: boolean;
+  closePDAsEnabled: boolean;
 }
 
 export class MonitorConfig extends CommonConfig {
@@ -82,6 +83,7 @@ export class MonitorConfig extends CommonConfig {
       BUNDLES_COUNT,
       BINANCE_WITHDRAW_WARN_THRESHOLD,
       BINANCE_WITHDRAW_ALERT_THRESHOLD,
+      CLOSE_PDAS_ENABLED,
     } = env;
 
     this.botModes = {
@@ -92,15 +94,14 @@ export class MonitorConfig extends CommonConfig {
       unknownRootBundleCallersEnabled: UNKNOWN_ROOT_BUNDLE_CALLERS_ENABLED === "true",
       stuckRebalancesEnabled: STUCK_REBALANCES_ENABLED === "true",
       spokePoolBalanceReportEnabled: REPORT_SPOKE_POOL_BALANCES === "true",
+      closePDAsEnabled: CLOSE_PDAS_ENABLED === "true",
       binanceWithdrawalLimitsEnabled:
         isDefined(BINANCE_WITHDRAW_WARN_THRESHOLD) || isDefined(BINANCE_WITHDRAW_ALERT_THRESHOLD),
     };
 
     // Used to monitor activities not from whitelisted data workers or relayers.
     this.whitelistedDataworkers = parseAddressesOptional(WHITELISTED_DATA_WORKERS);
     this.whitelistedRelayers = parseAddressesOptional(WHITELISTED_RELAYERS);
-
-    // Used to monitor balances, activities, etc. from the specified relayers.
     this.monitoredRelayers = parseAddressesOptional(MONITORED_RELAYERS);
     this.knownV1Addresses = parseAddressesOptional(KNOWN_V1_ADDRESSES);
     this.monitoredSpokePoolChains = JSON.parse(MONITORED_SPOKE_POOL_CHAINS ?? "[]");
@@ -211,5 +212,8 @@ export class MonitorConfig extends CommonConfig {
 
 const parseAddressesOptional = (addressJson?: string): Address[] => {
   const rawAddresses: string[] = addressJson ? JSON.parse(addressJson) : [];
-  return rawAddresses.map((address) => toAddressType(ethers.utils.getAddress(address), CHAIN_IDs.MAINNET));
+  return rawAddresses.map((address) => {
+    const chainId = address.startsWith("0x") ? CHAIN_IDs.MAINNET : CHAIN_IDs.SOLANA;
+    return toAddressType(address, chainId);
+  });
 };

src/monitor/index.ts

@@ -68,6 +68,12 @@ export async function runMonitor(_logger: winston.Logger, baseSigner: Signer): P
         logger.debug({ at: "Monitor#index", message: "Binance withdrawal limits check disabled" });
       }
 
+      if (config.botModes.closePDAsEnabled) {
+        await acrossMonitor.closePDAs();
+      } else {
+        logger.debug({ at: "Monitor#index", message: "Close PDAs disabled" });
+      }
+
       await clients.multiCallerClient.executeTxnQueues();
 
       logger.debug({ at: "Monitor#index", message: `Time to loop: ${(Date.now() - loopStart) / 1000}s` });

src/utils/BlockUtils.ts

@@ -28,7 +28,7 @@ export async function getBlockFinder(logger: winston.Logger, chainId: number): P
     return evmBlockFinders[chainId];
   }
   const provider = getSvmProvider(await getRedisCache());
-  svmBlockFinder ??= new SVMBlockFinder(logger, provider);
+  svmBlockFinder ??= new SVMBlockFinder(provider, [], logger);
   return svmBlockFinder;
 }
 
@@ -79,9 +79,13 @@ export async function getTimestampsForBundleStartBlocks(
           return [chainId, (await spokePoolClient.spokePool.getCurrentTime({ blockTag: startAt })).toNumber()];
         } else if (isSVMSpokePoolClient(spokePoolClient)) {
           const provider = spokePoolClient.svmEventsClient.getRpc();
-          const { timestamp } = await arch.svm.getNearestSlotTime(provider, spokePoolClient.logger, {
-            slot: BigInt(startAt),
-          });
+          const { timestamp } = await arch.svm.getNearestSlotTime(
+            provider,
+            {
+              slot: BigInt(startAt),
+            },
+            spokePoolClient.logger
+          );
           return [chainId, timestamp];
         }
       })

src/utils/FillUtils.ts

@@ -1,13 +1,33 @@
 import { HubPoolClient, SpokePoolClient } from "../clients";
-import { FillStatus, FillWithBlock, SpokePoolClientsByChain, DepositWithBlock } from "../interfaces";
-import { bnZero, CHAIN_IDs } from "../utils";
+import { FillStatus, FillWithBlock, SpokePoolClientsByChain, DepositWithBlock, RelayData } from "../interfaces";
+import { bnZero, CHAIN_IDs, EMPTY_MESSAGE } from "../utils";
 
 export type RelayerUnfilledDeposit = {
   deposit: DepositWithBlock;
   version: number;
   invalidFills: FillWithBlock[];
 };
 
+// @description Returns RelayData object with empty message.
+// @param fill  FillWithBlock object.
+// @returns RelayData object.
+export function getRelayDataFromFill(fill: FillWithBlock): RelayData {
+  return {
+    originChainId: fill.originChainId,
+    depositor: fill.depositor,
+    recipient: fill.recipient,
+    depositId: fill.depositId,
+    inputToken: fill.inputToken,
+    inputAmount: fill.inputAmount,
+    outputToken: fill.outputToken,
+    outputAmount: fill.outputAmount,
+    message: EMPTY_MESSAGE,
+    fillDeadline: fill.fillDeadline,
+    exclusiveRelayer: fill.exclusiveRelayer,
+    exclusivityDeadline: fill.exclusivityDeadline,
+  };
+}
+
 // @description Returns all unfilled deposits, indexed by destination chain.
 // @param destinationChainId  Chain ID to query outstanding deposits on.
 // @param spokePoolClients  Mapping of chainIds to SpokePoolClient objects.