Merge pull request #764 from ava-labs/throttle-account-txs

Throttle account txs

Author: cam-schultz

README.md

@@ -40,9 +40,6 @@ The most complicated example case that can arise above is that a feature depends
 
 Publishing all of the commits mentioned above to GitHub branches will enable running E2E tests through the CI.
 
-> [!TIP]
-> Running the tests locally doesn't require publishing the `subnet-evm` commit since `./scripts/e2e_test.sh` takes a flag specifying a local checkout of the `subnet-evm` repository.
-
 ## Releases
 GoReleaser is used to build the binaries of the services and also Docker images with those binaries. The monorepo feature of GoReleaser Pro is used to automate the release flow in response to tags like `signature-aggregator/v0.0.0`. The release actions in .github/workflows automate this, but the release build can also be run locally. Be sure to install the "pro" distribution of the command line utility, so that it can parse the `monorepo` key. For example:
 

messages/off-chain-registry/message_handler.go

@@ -192,7 +192,7 @@ func (m *messageHandler) SendMessage(
 		return common.Hash{}, err
 	}
 
-	txHash, err := m.destinationClient.SendTx(
+	receipt, err := m.destinationClient.SendTx(
 		signedMessage,
 		m.registryAddress.Hex(),
 		addProtocolVersionGasLimit,
@@ -206,7 +206,7 @@ func (m *messageHandler) SendMessage(
 		return common.Hash{}, err
 	}
 	m.logger.Info("Sent message to destination chain")
-	return txHash, nil
+	return receipt.TxHash, nil
 }
 
 func (m *messageHandler) LoggerWithContext(logger logging.Logger) logging.Logger {

messages/teleporter/message_handler.go

@@ -19,7 +19,6 @@ import (
 	"github.com/ava-labs/icm-services/messages"
 	pbDecider "github.com/ava-labs/icm-services/proto/pb/decider"
 	"github.com/ava-labs/icm-services/relayer/config"
-	"github.com/ava-labs/icm-services/utils"
 	"github.com/ava-labs/icm-services/vms"
 	"github.com/ava-labs/subnet-evm/accounts/abi/bind"
 	"github.com/ava-labs/subnet-evm/core/types"
@@ -289,7 +288,7 @@ func (m *messageHandler) SendMessage(
 		return common.Hash{}, err
 	}
 
-	txHash, err := m.destinationClient.SendTx(
+	receipt, err := m.destinationClient.SendTx(
 		signedMessage,
 		m.protocolAddress.Hex(),
 		gasLimit,
@@ -300,37 +299,34 @@ func (m *messageHandler) SendMessage(
 		return common.Hash{}, err
 	}
 
-	// Wait for the message to be included in a block before returning
-	receipt, err := m.waitForReceipt(signedMessage, m.destinationClient, txHash, m.teleporterMessageID)
-	if err != nil {
-		return common.Hash{}, err
-	}
+	txHash := receipt.TxHash
+
 	if receipt.Status != types.ReceiptStatusSuccessful {
 		// Check if the message has already been delivered to the destination chain
 		delivered, err := m.getTeleporterMessenger().MessageReceived(&bind.CallOpts{}, m.teleporterMessageID)
 		if err != nil {
 			m.logger.Error(
 				"Failed to check if message has been delivered to destination chain.",
 				zap.Error(err),
-				zap.String("txID", txHash.String()),
+				zap.Stringer("txID", txHash),
 			)
 			return common.Hash{}, fmt.Errorf("failed to check if message has been delivered: %w", err)
 		}
 		if delivered {
-			m.logger.Info("Execution reverted: message already delivered to destination.", zap.String("txID", txHash.String()))
+			m.logger.Info("Execution reverted: message already delivered to destination.", zap.Stringer("txID", txHash))
 			return txHash, nil
 		}
 
 		m.logger.Error(
 			"Transaction failed",
-			zap.String("txID", txHash.String()),
+			zap.Stringer("txID", txHash),
 		)
 		return common.Hash{}, fmt.Errorf("transaction failed with status: %d", receipt.Status)
 	}
 
 	m.logger.Info(
 		"Delivered message to destination chain",
-		zap.String("txID", txHash.String()),
+		zap.Stringer("txID", txHash),
 	)
 	return txHash, nil
 }
@@ -339,30 +335,6 @@ func (m *messageHandler) LoggerWithContext(logger logging.Logger) logging.Logger
 	return logger.With(m.logFields...)
 }
 
-func (m *messageHandler) waitForReceipt(
-	signedMessage *warp.Message,
-	destinationClient vms.DestinationClient,
-	txHash common.Hash,
-	teleporterMessageID ids.ID,
-) (*types.Receipt, error) {
-	var receipt *types.Receipt
-	operation := func() (err error) {
-		callCtx, callCtxCancel := context.WithTimeout(context.Background(), utils.DefaultRPCTimeout)
-		defer callCtxCancel()
-		receipt, err = destinationClient.Client().(ethclient.Client).TransactionReceipt(callCtx, txHash)
-		return err
-	}
-	err := utils.WithRetriesTimeout(m.logger, operation, destinationClient.TxInclusionTimeout(), "waitForReceipt")
-	if err != nil {
-		m.logger.Error(
-			"Failed to get transaction receipt",
-			zap.Error(err),
-		)
-		return nil, err
-	}
-	return receipt, nil
-}
-
 // parseTeleporterMessage returns the Warp message's corresponding Teleporter message from the cache if it exists.
 // Otherwise parses the Warp message payload.
 func (f *factory) parseTeleporterMessage(

messages/teleporter/message_handler_test.go

@@ -6,7 +6,6 @@ package teleporter
 import (
 	"math/big"
 	"testing"
-	"time"
 
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/utils/logging"
@@ -326,19 +325,10 @@ func TestSendMessageAlreadyDelivered(t *testing.T) {
 	mockClient.EXPECT().
 		Client().
 		Return(mockEthClient).
-		Times(2)
-
-	mockClient.EXPECT().
-		TxInclusionTimeout().
-		Return(time.Second * 1).
 		Times(1)
 
 	mockClient.EXPECT().
 		SendTx(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-		Times(1)
-
-	mockEthClient.EXPECT().
-		TransactionReceipt(gomock.Any(), gomock.Any()).
 		Return(
 			&types.Receipt{
 				Status: types.ReceiptStatusFailed,

relayer/README.md

@@ -436,18 +436,16 @@ If your temporary directory is not writable, the unit tests may fail with messag
 
 ### E2E Tests
 
-E2E tests are ran as part of CI, but can also be ran locally with the `--local` flag. To run the E2E tests locally, you'll need to install Gingko following the instructions [here](https://onsi.github.io/ginkgo/#installing-ginkgo)
-
-Next, provide the path to the `subnet-evm` repository and the path to a writeable data directory (this example uses `~/subnet-evm` and `~/tmp/e2e-test`) to use for the tests:
+To run the E2E tests locally, you'll need to install Gingko following the instructions [here](https://onsi.github.io/ginkgo/#installing-ginkgo). Run the tests using the dedicated script:
 
 ```bash
-./scripts/e2e_test.sh --local --subnet-evm ~/subnet-evm --data-dir ~/tmp/e2e-test
+./scripts/e2e_test.sh
 ```
 
 To run a specific E2E test, specify the environment variable `GINKGO_FOCUS`, which will then look for [test descriptions](./tests/e2e_test.go#L68) that match the provided input. For example, to run the `Basic Relay` test:
 
 ```bash
-GINKGO_FOCUS="Basic" ./scripts/e2e_test.sh --local --subnet-evm ~/subnet-evm --data-dir ~/tmp/e2e-test
+GINKGO_FOCUS="Basic" ./scripts/e2e_test.sh
 ```
 
 The E2E tests use the `TeleporterMessenger` contract deployment transaction specified in the following files:

relayer/api/relay_message.go

@@ -63,11 +63,22 @@ func relayMessageAPIHandler(logger logging.Logger, messageCoordinator *relayer.M
 			return
 		}
 
+		if !common.IsHexAddress(req.SourceAddress) {
+			logger.Warn("Invalid source address")
+			http.Error(w, "invalid source address", http.StatusBadRequest)
+			return
+		}
+		address := common.HexToAddress(req.SourceAddress)
+
 		warpMessageInfo := &types.WarpMessageInfo{
-			SourceAddress:   common.HexToAddress(req.SourceAddress),
+			SourceAddress:   address,
 			UnsignedMessage: unsignedMessage,
 		}
-
+		logger.Info(
+			"Processing manual warp message",
+			zap.Stringer("sourceAddress", address),
+			zap.Stringer("messageID", unsignedMessage.ID()),
+		)
 		txHash, err := messageCoordinator.ProcessWarpMessage(warpMessageInfo)
 		if err != nil {
 			logger.Error("Error processing message", zap.Error(err))

scripts/test.sh

@@ -37,4 +37,4 @@ go build -o tests/cmd/decider/decider ./tests/cmd/decider/
 
 "$root"/scripts/generate.sh
 
-go test -tags testing $VERBOSE ./...
+go test -tags testing -timeout 30s $VERBOSE ./...

vms/destination_client.go

@@ -7,13 +7,13 @@ package vms
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/vms/platformvm/warp"
 	"github.com/ava-labs/icm-services/relayer/config"
 	"github.com/ava-labs/icm-services/vms/evm"
+	"github.com/ava-labs/subnet-evm/core/types"
 	"github.com/ethereum/go-ethereum/common"
 	"go.uber.org/zap"
 )
@@ -25,7 +25,7 @@ type DestinationClient interface {
 	// SendTx constructs the transaction from warp primitives, and sends to the configured destination chain endpoint.
 	// Returns the hash of the sent transaction.
 	// TODO: Make generic for any VM.
-	SendTx(signedMessage *warp.Message, toAddress string, gasLimit uint64, callData []byte) (common.Hash, error)
+	SendTx(signedMessage *warp.Message, toAddress string, gasLimit uint64, callData []byte) (*types.Receipt, error)
 
 	// Client returns the underlying client for the destination chain
 	Client() interface{}
@@ -38,9 +38,6 @@ type DestinationClient interface {
 
 	// BlockGasLimit returns destination blockchain block gas limit
 	BlockGasLimit() uint64
-
-	// TxInclusionTimeout returns the timeout for waiting for a transaction to be included on the destination chain
-	TxInclusionTimeout() time.Duration
 }
 
 func NewDestinationClient(