feat(transactor): More accurate tracking of mempool upon waiting for tx confirmation

client/eth/client.go

@@ -52,21 +52,21 @@ type Reader interface {
 
 		{
 			"pending": {
-				"1": {
+				1: {
 					// transaction details...
 				},
 				...
 			},
 			"queued": {
-				"3": {
+				3: {
 					// transaction details...
 				},
 				...
 			}
 		}
 	*/
 	TxPoolContentFrom(ctx context.Context, address common.Address) (
-		map[string]map[string]*ethcoretypes.Transaction, error,
+		map[string]map[uint64]*ethcoretypes.Transaction, error,
 	)
 
 	/*
@@ -76,19 +76,19 @@ type Reader interface {
 		{
 			"pending": {
 				"0x12345": {
-					"1": "0x12345789: 1 wei + 2 gas x 3 wei"
+					1: "0x12345789: 1 wei + 2 gas x 3 wei"
 				},
 				...
 			},
 			"queued": {
 				"0x67890": {
-					"2": "0x12345789: 1 wei + 2 gas x 3 wei"
+					2: "0x12345789: 1 wei + 2 gas x 3 wei"
 				},
 				...
 			}
 		}
 	*/
-	TxPoolInspect(ctx context.Context) (map[string]map[common.Address]map[string]string, error)
+	TxPoolInspect(ctx context.Context) (map[string]map[common.Address]map[uint64]string, error)
 }
 
 type Writer interface {
@@ -129,7 +129,7 @@ func (c *ExtendedEthClient) Close() error {
 	if c == nil {
 		return ErrClosed
 	}
-	c.Close()
+	c.Client.Close()
 	return nil
 }
 
@@ -180,8 +180,8 @@ func (c *ExtendedEthClient) SubscribeFilterLogs(
 
 func (c *ExtendedEthClient) TxPoolContentFrom(
 	ctx context.Context, address common.Address,
-) (map[string]map[string]*ethcoretypes.Transaction, error) {
-	var result map[string]map[string]*ethcoretypes.Transaction
+) (map[string]map[uint64]*ethcoretypes.Transaction, error) {
+	var result map[string]map[uint64]*ethcoretypes.Transaction
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, c.rpcTimeout)
 	defer cancel()
 	if err := c.Client.Client().CallContext(
@@ -194,8 +194,8 @@ func (c *ExtendedEthClient) TxPoolContentFrom(
 
 func (c *ExtendedEthClient) TxPoolInspect(
 	ctx context.Context,
-) (map[string]map[common.Address]map[string]string, error) {
-	var result map[string]map[common.Address]map[string]string
+) (map[string]map[common.Address]map[uint64]string, error) {
+	var result map[string]map[common.Address]map[uint64]string
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, c.rpcTimeout)
 	defer cancel()
 	if err := c.Client.Client().CallContext(

client/eth/client_provider.go

@@ -261,21 +261,21 @@ Example response:
 
 	{
 		"pending": {
-			"1": {
+			1: {
 				// transaction details...
 			},
 			...
 		},
 		"queued": {
-			"3": {
+			3: {
 				// transaction details...
 			},
 			...
 		}
 	}
 */
 func (c *ChainProviderImpl) TxPoolContentFrom(ctx context.Context, address common.Address) (
-	map[string]map[string]*types.Transaction, error,
+	map[string]map[uint64]*types.Transaction, error,
 ) {
 	if client, ok := c.GetHTTP(); ok {
 		ctxWithTimeout, cancel := context.WithTimeout(ctx, c.rpcTimeout)
@@ -292,20 +292,20 @@ Example response:
 	{
 		"pending": {
 			"0x12345": {
-				"1": "0x12345789: 1 wei + 2 gas x 3 wei"
+				1: "0x12345789: 1 wei + 2 gas x 3 wei"
 			},
 			...
 		},
 		"queued": {
 			"0x67890": {
-				"2": "0x12345789: 1 wei + 2 gas x 3 wei"
+				2: "0x12345789: 1 wei + 2 gas x 3 wei"
 			},
 			...
 		}
 	}
 */
 func (c *ChainProviderImpl) TxPoolInspect(ctx context.Context) (
-	map[string]map[common.Address]map[string]string, error,
+	map[string]map[common.Address]map[uint64]string, error,
 ) {
 	if client, ok := c.GetHTTP(); ok {
 		ctxWithTimeout, cancel := context.WithTimeout(ctx, c.rpcTimeout)

client/eth/client_test.go

@@ -0,0 +1,156 @@
+package eth_test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/berachain/offchain-sdk/client/eth"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	ethcoretypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+// This file tests the methods on the extended eth client.
+//
+// Note that the following must be set up for these tests to run:
+//  - the chain RPC is available at env var `ETH_RPC_URL`
+//  - OR the chain WS RPC is available at env var `ETH_RPC_URL_WS`
+//  - the RPC endpoint must return within 5 seconds or tests will timeout
+//  - [Optional for `txPoolContentFrom`] a wallet to query the mempool for at env var `ETH_ADDR`
+
+const (
+	TestModeHTTP int = iota
+	TestModeWS
+	TestModeEither
+)
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL` or `ETH_RPC_URL_WS`.
+func setUp(testMode int, t *testing.T) (*eth.ExtendedEthClient, error) {
+	rpcTimeout := 5 * time.Second
+	ctxWithTimeout, cancel := context.WithTimeout(context.Background(), rpcTimeout)
+	defer cancel()
+
+	var ethRPC string
+	switch testMode {
+	case TestModeWS:
+		ethRPC = os.Getenv("ETH_RPC_URL_WS")
+	case TestModeHTTP:
+		ethRPC = os.Getenv("ETH_RPC_URL")
+	case TestModeEither:
+		if ethRPC = os.Getenv("ETH_RPC_URL_WS"); ethRPC == "" {
+			ethRPC = os.Getenv("ETH_RPC_URL")
+		}
+	default:
+		panic("invalid test mode")
+	}
+	if ethRPC == "" {
+		t.Skipf("Skipping test: no eth rpc url provided")
+	}
+
+	ethClient, err := ethclient.DialContext(ctxWithTimeout, ethRPC)
+	if err != nil {
+		return nil, err
+	}
+
+	eec := eth.NewExtendedEthClient(ethClient, rpcTimeout)
+	return eec, nil
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL` or `ETH_RPC_URL_WS`.
+func TestClose(t *testing.T) {
+	eec, err := setUp(TestModeEither, t)
+	assert.NoError(t, err)
+
+	err = eec.Close()
+	assert.NoError(t, err)
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL` or `ETH_RPC_URL_WS`.
+func TestHealth(t *testing.T) {
+	eec, err := setUp(TestModeEither, t)
+	assert.NoError(t, err)
+
+	health := eec.Health()
+	assert.True(t, health)
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL`.
+func TestGetReceipts(t *testing.T) {
+	eec, err := setUp(TestModeHTTP, t)
+	assert.NoError(t, err)
+
+	ctx := context.Background()
+	txs := ethcoretypes.Transactions{}
+
+	receipts, err := eec.GetReceipts(ctx, txs)
+	assert.NoError(t, err)
+	assert.Empty(t, receipts)
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL_WS`.
+func TestSubscribeNewHead(t *testing.T) {
+	eec, err := setUp(TestModeWS, t)
+	assert.NoError(t, err)
+
+	ctx := context.Background()
+
+	ch, sub, err := eec.SubscribeNewHead(ctx)
+	assert.NoError(t, err)
+	assert.NotNil(t, ch)
+	assert.NotNil(t, sub)
+
+	assert.NotPanics(t, func() { sub.Unsubscribe() })
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL_WS`.
+func TestSubscribeFilterLogs(t *testing.T) {
+	eec, err := setUp(TestModeWS, t)
+	assert.NoError(t, err)
+
+	ctx := context.Background()
+	query := ethereum.FilterQuery{}
+	ch := make(chan ethcoretypes.Log)
+
+	sub, err := eec.SubscribeFilterLogs(ctx, query, ch)
+	assert.NoError(t, err)
+	assert.NotNil(t, sub)
+
+	assert.NotPanics(t, func() { sub.Unsubscribe() })
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL` AND a wallet to query the
+// txpool for at `ETH_ADDR`.
+func TestTxPoolContentFrom(t *testing.T) {
+	eec, err := setUp(TestModeHTTP, t)
+	assert.NoError(t, err)
+
+	ctx := context.Background()
+	addrStr := os.Getenv("ETH_ADDR")
+	if addrStr == "" {
+		t.Skipf("Skipping test: no eth address provided")
+	}
+	address := common.HexToAddress(addrStr)
+
+	result, err := eec.TxPoolContentFrom(ctx, address)
+	assert.NoError(t, err)
+	assert.NotNil(t, result)
+
+	t.Log("result", result)
+}
+
+// NOTE: requires Ethereum chain rpc url at env var `ETH_RPC_URL`.
+func TestTxPoolInspect(t *testing.T) {
+	eec, err := setUp(TestModeHTTP, t)
+	assert.NoError(t, err)
+
+	ctx := context.Background()
+
+	result, err := eec.TxPoolInspect(ctx)
+	assert.NoError(t, err)
+	assert.NotNil(t, result)
+}

core/transactor/config.go

@@ -16,16 +16,16 @@ type Config struct {
 	WaitFullBatchTimeout bool
 	// How long to wait to retrieve txs from the queue if it is empty (ideally quick <= 1s).
 	EmptyQueueDelay time.Duration
+	// What the `requireSuccess` flag should be set for calls to `multicall`, if batching txs.
+	MulticallRequireSuccess bool
 
 	// Maximum duration allowed for the tx to be signed (increase this if using a remote signer)
 	SignTxTimeout time.Duration
 
 	// How long to wait for the pending nonce (ideally 1 block time).
 	PendingNonceInterval time.Duration
-	// How long to wait for a tx to hit the mempool (ideally 1-2 block time).
-	InMempoolTimeout time.Duration
-	// How long to wait for a tx to be mined/confirmed by the chain.
-	TxReceiptTimeout time.Duration
+	// How long to wait for a tx to hit the mempool and/or be confirmed by the chain.
+	TxWaitingTimeout time.Duration
 	// Whether we should resend txs that are stale (not confirmed after the receipt timeout).
 	ResendStaleTxs bool
 

core/transactor/factory/batcher/multicall3.go

@@ -39,7 +39,9 @@ func NewMulticall3(address common.Address) *Multicall3 {
 }
 
 // BatchRequests creates a batched transaction request for the given call requests.
-func (mc *Multicall3) BatchRequests(callReqs ...*ethereum.CallMsg) *types.Request {
+func (mc *Multicall3) BatchRequests(
+	requireSuccess bool, callReqs ...*ethereum.CallMsg,
+) *types.Request {
 	var (
 		calls       = make([]bindings.Multicall3Call, len(callReqs))
 		totalValue  = big.NewInt(0)
@@ -73,20 +75,20 @@ func (mc *Multicall3) BatchRequests(callReqs ...*ethereum.CallMsg) *types.Reques
 
 	txRequest, _ := mc.packer.CreateRequest(
 		"", mc.contractAddress, totalValue, gasTipCap, gasFeeCap, gasLimit,
-		tryAggregate, false, calls,
+		tryAggregate, requireSuccess, calls,
 	)
 	return txRequest
 }
 
 // BatchCallRequests uses the Multicall3 contract to create a batched call request for the given
 // call messages and return the batched call result data for each call, as a `[]Multicall3Result`.
 func (mc *Multicall3) BatchCallRequests(
-	ctx context.Context, from common.Address, callReqs ...*ethereum.CallMsg,
+	ctx context.Context, from common.Address, requireSuccess bool, callReqs ...*ethereum.CallMsg,
 ) (any, error) {
 	sCtx := sdk.UnwrapContext(ctx)
 
 	// get the batched tx (call) requests
-	batchedCall := mc.BatchRequests(callReqs...)
+	batchedCall := mc.BatchRequests(requireSuccess, callReqs...)
 	batchedCall.From = from
 
 	// call the multicall3 contract with the batched call request

core/transactor/factory/batcher/multicall3_test.go

@@ -46,7 +46,7 @@ func TestMulticall3(t *testing.T) {
 	}
 
 	// batch and send the calls to the chain
-	resp, err := multicaller.BatchCallRequests(sCtx, empty, call1.CallMsg, call2.CallMsg)
+	resp, err := multicaller.BatchCallRequests(sCtx, empty, false, call1.CallMsg, call2.CallMsg)
 	if err != nil {
 		t.Fatal(err)
 	}