Fix tx tracker performance issues (#537)

### Description

The `tx-tracker` service is failing to process messages at the speed we need. This pull request is an attempt at fixing the issue.

deploy/tx-tracker/env/production.env

@@ -12,66 +12,66 @@ SQS_AWS_REGION=
 AWS_IAM_ROLE=
 
 ACALA_BASE_URL=https://eth-rpc-acala.aca-api.network
-ACALA_REQUESTS_PER_MINUTE=4
+ACALA_REQUESTS_PER_MINUTE=12
 
 ALGORAND_BASE_URL=https://mainnet-idx.algonode.cloud
-ALGORAND_REQUESTS_PER_MINUTE=4
+ALGORAND_REQUESTS_PER_MINUTE=12
 
 APTOS_BASE_URL=https://fullnode.mainnet.aptoslabs.com/v1
-APTOS_REQUESTS_PER_MINUTE=8
+APTOS_REQUESTS_PER_MINUTE=12
 
 ARBITRUM_BASE_URL=https://rpc.ankr.com/arbitrum
-ARBITRUM_REQUESTS_PER_MINUTE=4
+ARBITRUM_REQUESTS_PER_MINUTE=12
 
 AVALANCHE_BASE_URL=https://api.avax.network/ext/bc/C/rpc
-AVALANCHE_REQUESTS_PER_MINUTE=6
+AVALANCHE_REQUESTS_PER_MINUTE=12
 
 BSC_BASE_URL=https://bsc-dataseed2.defibit.io
-BSC_REQUESTS_PER_MINUTE=6
+BSC_REQUESTS_PER_MINUTE=12
 
 CELO_BASE_URL=https://forno.celo.org
-CELO_REQUESTS_PER_MINUTE=4
+CELO_REQUESTS_PER_MINUTE=12
 
 ETHEREUM_BASE_URL=https://rpc.ankr.com/eth
-ETHEREUM_REQUESTS_PER_MINUTE=6
+ETHEREUM_REQUESTS_PER_MINUTE=12
 
 FANTOM_BASE_URL=https://rpc.ankr.com/fantom
-FANTOM_REQUESTS_PER_MINUTE=4
+FANTOM_REQUESTS_PER_MINUTE=12
 
 # Currently, the main URL for injective RPC (https://k8s.mainnet.lcd.injective.network) is down.
 # We're temporally using a different one.
 INJECTIVE_BASE_URL=http://sentry2.injective.network:10337/
-INJECTIVE_REQUESTS_PER_MINUTE=4
+INJECTIVE_REQUESTS_PER_MINUTE=12
 
 KARURA_BASE_URL=https://eth-rpc-karura.aca-api.network
-KARURA_REQUESTS_PER_MINUTE=4
+KARURA_REQUESTS_PER_MINUTE=12
 
 KLAYTN_BASE_URL=https://klaytn-mainnet-rpc.allthatnode.com:8551
-KLAYTN_REQUESTS_PER_MINUTE=4
+KLAYTN_REQUESTS_PER_MINUTE=12
 
 MOONBEAM_BASE_URL=https://rpc.api.moonbeam.network
-MOONBEAM_REQUESTS_PER_MINUTE=4
+MOONBEAM_REQUESTS_PER_MINUTE=12
 
 OASIS_BASE_URL=https://emerald.oasis.dev
-OASIS_REQUESTS_PER_MINUTE=4
+OASIS_REQUESTS_PER_MINUTE=12
 
 OPTIMISM_BASE_URL=https://rpc.ankr.com/optimism
-OPTIMISM_REQUESTS_PER_MINUTE=4
+OPTIMISM_REQUESTS_PER_MINUTE=12
 
 POLYGON_BASE_URL=https://rpc.ankr.com/polygon
-POLYGON_REQUESTS_PER_MINUTE=6
+POLYGON_REQUESTS_PER_MINUTE=12
 
 SOLANA_BASE_URL=https://api.mainnet-beta.solana.com
-SOLANA_REQUESTS_PER_MINUTE=8
+SOLANA_REQUESTS_PER_MINUTE=12
 
 SUI_BASE_URL=https://fullnode.mainnet.sui.io:443
-SUI_REQUESTS_PER_MINUTE=4
+SUI_REQUESTS_PER_MINUTE=12
 
 TERRA_BASE_URL=https://columbus-fcd.terra.dev
-TERRA_REQUESTS_PER_MINUTE=4
+TERRA_REQUESTS_PER_MINUTE=12
 
 TERRA2_BASE_URL=https://phoenix-lcd.terra.dev
-TERRA2_REQUESTS_PER_MINUTE=4
+TERRA2_REQUESTS_PER_MINUTE=12
 
 XPLA_BASE_URL=https://dimension-lcd.xpla.dev
-XPLA_REQUESTS_PER_MINUTE=4
+XPLA_REQUESTS_PER_MINUTE=12

deploy/tx-tracker/env/staging.env

@@ -12,66 +12,66 @@ SQS_AWS_REGION=
 AWS_IAM_ROLE=
 
 ACALA_BASE_URL=https://eth-rpc-acala.aca-api.network
-ACALA_REQUESTS_PER_MINUTE=2
+ACALA_REQUESTS_PER_MINUTE=12
 
 ALGORAND_BASE_URL=https://mainnet-idx.algonode.cloud
-ALGORAND_REQUESTS_PER_MINUTE=2
+ALGORAND_REQUESTS_PER_MINUTE=12
 
 APTOS_BASE_URL=https://fullnode.mainnet.aptoslabs.com/v1
 APTOS_REQUESTS_PER_MINUTE=12
 
 ARBITRUM_BASE_URL=https://rpc.ankr.com/arbitrum
-ARBITRUM_REQUESTS_PER_MINUTE=2
+ARBITRUM_REQUESTS_PER_MINUTE=12
 
 AVALANCHE_BASE_URL=https://api.avax.network/ext/bc/C/rpc
-AVALANCHE_REQUESTS_PER_MINUTE=8
+AVALANCHE_REQUESTS_PER_MINUTE=12
 
 BSC_BASE_URL=https://bsc-dataseed2.defibit.io
-BSC_REQUESTS_PER_MINUTE=8
+BSC_REQUESTS_PER_MINUTE=12
 
 CELO_BASE_URL=https://forno.celo.org
-CELO_REQUESTS_PER_MINUTE=2
+CELO_REQUESTS_PER_MINUTE=12
 
 ETHEREUM_BASE_URL=https://rpc.ankr.com/eth
-ETHEREUM_REQUESTS_PER_MINUTE=8
+ETHEREUM_REQUESTS_PER_MINUTE=12
 
 FANTOM_BASE_URL=https://rpc.ankr.com/fantom
-FANTOM_REQUESTS_PER_MINUTE=2
+FANTOM_REQUESTS_PER_MINUTE=12
 
 # Currently, the main URL for injective RPC (https://k8s.mainnet.lcd.injective.network) is down.
 # We're temporally using a different one.
 INJECTIVE_BASE_URL=http://sentry2.injective.network:10337/
-INJECTIVE_REQUESTS_PER_MINUTE=2
+INJECTIVE_REQUESTS_PER_MINUTE=12
 
 KARURA_BASE_URL=https://eth-rpc-karura.aca-api.network
-KARURA_REQUESTS_PER_MINUTE=2
+KARURA_REQUESTS_PER_MINUTE=12
 
 KLAYTN_BASE_URL=https://klaytn-mainnet-rpc.allthatnode.com:8551
-KLAYTN_REQUESTS_PER_MINUTE=2
+KLAYTN_REQUESTS_PER_MINUTE=12
 
 MOONBEAM_BASE_URL=https://rpc.api.moonbeam.network
-MOONBEAM_REQUESTS_PER_MINUTE=2
+MOONBEAM_REQUESTS_PER_MINUTE=12
 
 OASIS_BASE_URL=https://emerald.oasis.dev
-OASIS_REQUESTS_PER_MINUTE=2
+OASIS_REQUESTS_PER_MINUTE=12
 
 OPTIMISM_BASE_URL=https://rpc.ankr.com/optimism
-OPTIMISM_REQUESTS_PER_MINUTE=2
+OPTIMISM_REQUESTS_PER_MINUTE=12
 
 POLYGON_BASE_URL=https://rpc.ankr.com/polygon
-POLYGON_REQUESTS_PER_MINUTE=8
+POLYGON_REQUESTS_PER_MINUTE=12
 
 SOLANA_BASE_URL=https://api.mainnet-beta.solana.com
 SOLANA_REQUESTS_PER_MINUTE=12
 
 SUI_BASE_URL=https://fullnode.mainnet.sui.io:443
-SUI_REQUESTS_PER_MINUTE=2
+SUI_REQUESTS_PER_MINUTE=12
 
 TERRA_BASE_URL=https://columbus-fcd.terra.dev
-TERRA_REQUESTS_PER_MINUTE=2
+TERRA_REQUESTS_PER_MINUTE=12
 
 TERRA2_BASE_URL=https://phoenix-lcd.terra.dev
-TERRA2_REQUESTS_PER_MINUTE=2
+TERRA2_REQUESTS_PER_MINUTE=12
 
 XPLA_BASE_URL=https://dimension-lcd.xpla.dev
-XPLA_REQUESTS_PER_MINUTE=2
+XPLA_REQUESTS_PER_MINUTE=12

deploy/tx-tracker/env/test.env

@@ -12,64 +12,64 @@ SQS_AWS_REGION=
 AWS_IAM_ROLE=
 
 ACALA_BASE_URL=https://acala-dev.aca-dev.network/eth/http
-ACALA_REQUESTS_PER_MINUTE=2
+ACALA_REQUESTS_PER_MINUTE=12
 
 ALGORAND_BASE_URL=https://testnet-idx.algonode.cloud
-ALGORAND_REQUESTS_PER_MINUTE=2
+ALGORAND_REQUESTS_PER_MINUTE=12
 
 APTOS_BASE_URL=https://fullnode.testnet.aptoslabs.com/v1
-APTOS_REQUESTS_PER_MINUTE=2
+APTOS_REQUESTS_PER_MINUTE=12
 
 ARBITRUM_BASE_URL=https://goerli-rollup.arbitrum.io/rpc
-ARBITRUM_REQUESTS_PER_MINUTE=2
+ARBITRUM_REQUESTS_PER_MINUTE=12
 
 AVALANCHE_BASE_URL=https://rpc.ankr.com/avalanche_fuji
-AVALANCHE_REQUESTS_PER_MINUTE=2
+AVALANCHE_REQUESTS_PER_MINUTE=12
 
 BSC_BASE_URL=https://data-seed-prebsc-1-s1.binance.org:8545
-BSC_REQUESTS_PER_MINUTE=2
+BSC_REQUESTS_PER_MINUTE=12
 
 CELO_BASE_URL=https://alfajores-forno.celo-testnet.org
-CELO_REQUESTS_PER_MINUTE=2
+CELO_REQUESTS_PER_MINUTE=12
 
 ETHEREUM_BASE_URL=https://rpc.ankr.com/eth_goerli
-ETHEREUM_REQUESTS_PER_MINUTE=2
+ETHEREUM_REQUESTS_PER_MINUTE=12
 
 FANTOM_BASE_URL=https://rpc.testnet.fantom.network
-FANTOM_REQUESTS_PER_MINUTE=2
+FANTOM_REQUESTS_PER_MINUTE=12
 
 INJECTIVE_BASE_URL=https://k8s.testnet.tm.injective.network:443
-INJECTIVE_REQUESTS_PER_MINUTE=2
+INJECTIVE_REQUESTS_PER_MINUTE=12
 
 KARURA_BASE_URL=https://karura-dev.aca-dev.network/eth/http
-KARURA_REQUESTS_PER_MINUTE=2
+KARURA_REQUESTS_PER_MINUTE=12
 
 KLAYTN_BASE_URL=https://api.baobab.klaytn.net:8651
-KLAYTN_REQUESTS_PER_MINUTE=2
+KLAYTN_REQUESTS_PER_MINUTE=12
 
 MOONBEAM_BASE_URL=https://rpc.api.moonbase.moonbeam.network
-MOONBEAM_REQUESTS_PER_MINUTE=2
+MOONBEAM_REQUESTS_PER_MINUTE=12
 
 OASIS_BASE_URL=https://testnet.emerald.oasis.dev
-OASIS_REQUESTS_PER_MINUTE=2
+OASIS_REQUESTS_PER_MINUTE=12
 
 OPTIMISM_BASE_URL=https://goerli.optimism.io
-OPTIMISM_REQUESTS_PER_MINUTE=2
+OPTIMISM_REQUESTS_PER_MINUTE=12
 
 POLYGON_BASE_URL=https://rpc.ankr.com/polygon_mumbai
-POLYGON_REQUESTS_PER_MINUTE=2
+POLYGON_REQUESTS_PER_MINUTE=12
 
 SOLANA_BASE_URL=https://api.devnet.solana.com
-SOLANA_REQUESTS_PER_MINUTE=2
+SOLANA_REQUESTS_PER_MINUTE=12
 
 SUI_BASE_URL=https://fullnode.testnet.sui.io:443
-SUI_REQUESTS_PER_MINUTE=2
+SUI_REQUESTS_PER_MINUTE=12
 
 TERRA_BASE_URL=https://bombay-lcd.terra.dev
-TERRA_REQUESTS_PER_MINUTE=4
+TERRA_REQUESTS_PER_MINUTE=12
 
 TERRA2_BASE_URL=https://pisco-lcd.terra.dev
-TERRA2_REQUESTS_PER_MINUTE=2
+TERRA2_REQUESTS_PER_MINUTE=12
 
 XPLA_BASE_URL=https://cube-lcd.xpla.dev:443
-XPLA_REQUESTS_PER_MINUTE=2
+XPLA_REQUESTS_PER_MINUTE=12

tx-tracker/chains/api_cosmos.go

@@ -42,9 +42,10 @@ func fetchCosmosTx(
  // Perform the HTTP request
  uri := fmt.Sprintf("%s/cosmos/tx/v1beta1/txs/%s", baseUrl, txHash)
  body, err := httpGet(ctx, rateLimiter, uri)
- if strings.Contains(err.Error(), "404") {
- return nil, ErrTransactionNotFound
- } else if err != nil {
+ if err != nil {
+ if strings.Contains(err.Error(), "404") {
+ return nil, ErrTransactionNotFound
+ }
  return nil, fmt.Errorf("failed to query cosmos tx endpoint: %w", err)
  }
 

tx-tracker/chains/api_sui.go

@@ -53,9 +53,10 @@ func fetchSuiTx(
  // Execute the remote procedure call
  opts := suiGetTransactionBlockOpts{ShowInput: true}
  err = client.CallContext(ctx, &reply, "sui_getTransactionBlock", txHash, opts)
- if strings.Contains(err.Error(), "Could not find the referenced transaction") {
- return nil, ErrTransactionNotFound
- } else if err != nil {
+ if err != nil {
+ if strings.Contains(err.Error(), "Could not find the referenced transaction") {
+ return nil, ErrTransactionNotFound
+ }
  return nil, fmt.Errorf("failed to get tx by hash: %w", err)
  }
  }

tx-tracker/consumer/consumer.go

@@ -55,30 +55,19 @@ func (c *Consumer) producerLoop(ctx context.Context) {
 
 func (c *Consumer) process(ctx context.Context, msg queue.ConsumerMessage) {
 
+ defer msg.Done()
+
  event := msg.Data()
 
  // Do not process messages from PythNet
  if event.ChainID == sdk.ChainIDPythNet {
- if !msg.IsExpired() {
- c.logger.Debug("Deleting PythNet message", zap.String("vaaId", event.ID))
- msg.Done()
- } else {
- c.logger.Debug("Skipping expired PythNet message", zap.String("vaaId", event.ID))
- }
- return
- }
-
- // Skip non-processed, expired messages
- if msg.IsExpired() {
- c.logger.Warn("Message expired - skipping",
- zap.String("vaaId", event.ID),
- zap.Bool("isExpired", msg.IsExpired()),
- )
+ c.logger.Debug("Skipping expired PythNet message", zap.String("vaaId", event.ID))
  return
  }
 
  // Process the VAA
  p := ProcessSourceTxParams{
+ Timestamp: event.Timestamp,
  VaaId: event.ID,
  ChainId: event.ChainID,
  Emitter: event.EmitterAddress,
@@ -97,27 +86,14 @@ func (c *Consumer) process(ctx context.Context, msg queue.ConsumerMessage) {
  c.logger.Warn("Message already processed - skipping",
  zap.String("vaaId", event.ID),
  )
- } else if errors.Is(err, chains.ErrTransactionNotFound) {
- c.logger.Warn("Transaction not found - will retry after SQS visibilityTimeout",
- zap.String("vaaId", event.ID),
- )
- return
  } else if err != nil {
- c.logger.Error("Failed to process originTx - will retry after SQS visibilityTimeout",
+ c.logger.Error("Failed to process originTx",
  zap.String("vaaId", event.ID),
  zap.Error(err),
  )
- return
  } else {
  c.logger.Info("Transaction processed successfully",
  zap.String("id", event.ID),
  )
  }
-
- // Mark the message as done
- //
- // If the message is expired, it will be put back into the queue.
- if !msg.IsExpired() {
- msg.Done()
- }
 }