Txpool pending (#266)

* re-encode transaction count request and send to sequencer for RPC nodes

* nicer implementation of the marshalling for block number

* only count pending transactions for nonce check in the pool

* forward txpool calls to sequencer from RPC node

* adding comment around why we ignore queued transactions when calcing the nonce

cmd/rpcdaemon/commands/daemon.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ledgerwatch/erigon/turbo/rpchelper"
 	"github.com/ledgerwatch/erigon/turbo/services"
 	"github.com/ledgerwatch/erigon/zk/syncer"
+	"github.com/ledgerwatch/erigon/zk/sequencer"
 )
 
 // APIList describes the list of available RPC apis
@@ -21,11 +22,18 @@ func APIList(db kv.RoDB, borDb kv.RoDB, eth rpchelper.ApiBackend, txPool txpool.
 	blockReader services.FullBlockReader, agg *libstate.AggregatorV3, cfg httpcfg.HttpCfg, engine consensus.EngineReader,
 	zkConfig *ethconfig.Zk, l1Syncer *syncer.L1Syncer,
 ) (list []rpc.API) {
+
+	// non-sequencer nodes should forward on requests to the sequencer
+	rpcUrl := ""
+	if !sequencer.IsSequencer() {
+		rpcUrl = zkConfig.L2RpcUrl
+	}
+
 	base := NewBaseApi(filters, stateCache, blockReader, agg, cfg.WithDatadir, cfg.EvmCallTimeout, engine, cfg.Dirs)
 	base.SetL2RpcUrl(zkConfig.L2RpcUrl)
 	ethImpl := NewEthAPI(base, db, eth, txPool, mining, cfg.Gascap, cfg.ReturnDataLimit, zkConfig)
 	erigonImpl := NewErigonAPI(base, db, eth)
-	txpoolImpl := NewTxPoolAPI(base, db, txPool)
+	txpoolImpl := NewTxPoolAPI(base, db, txPool, rpcUrl)
 	netImpl := NewNetAPIImpl(eth)
 	debugImpl := NewPrivateDebugAPI(base, db, cfg.Gascap)
 	traceImpl := NewTraceAPI(base, db, &cfg)

cmd/rpcdaemon/commands/eth_accounts.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ledgerwatch/erigon/common"
 	"github.com/ledgerwatch/erigon/common/hexutil"
 	"github.com/ledgerwatch/erigon/rpc"
+	"github.com/ledgerwatch/erigon/zk/sequencer"
 )
 
 // GetBalance implements eth_getBalance. Returns the balance of an account for a given address.
@@ -45,6 +46,15 @@ func (api *APIImpl) GetBalance(ctx context.Context, address libcommon.Address, b
 
 // GetTransactionCount implements eth_getTransactionCount. Returns the number of transactions sent from an address (the nonce).
 func (api *APIImpl) GetTransactionCount(ctx context.Context, address libcommon.Address, blockNrOrHash *rpc.BlockNumberOrHash) (*hexutil.Uint64, error) {
+	// zkevm: forward requests to the sequencer
+	if !sequencer.IsSequencer() {
+		res, err := api.sendGetTransactionCountToSequencer(api.l2RpcUrl, address, blockNrOrHash)
+		if err != nil {
+			return nil, err
+		}
+		return res, nil
+	}
+
 	// if not set, use latest
 	if blockNrOrHash == nil {
 		tmp := rpc.BlockNumberOrHashWithNumber(rpc.LatestBlockNumber)

cmd/rpcdaemon/commands/eth_accounts_zk.go

@@ -0,0 +1,47 @@
+package commands
+
+import (
+	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
+	"fmt"
+	"github.com/ledgerwatch/erigon/common/hexutil"
+	libcommon "github.com/gateway-fm/cdk-erigon-lib/common"
+	"github.com/ledgerwatch/erigon/rpc"
+	"strings"
+	"github.com/ledgerwatch/erigon/zkevm/hex"
+)
+
+func (api *APIImpl) sendGetTransactionCountToSequencer(rpcUrl string, address libcommon.Address, blockNrOrHash *rpc.BlockNumberOrHash) (*hexutil.Uint64, error) {
+	addressHex := "0x" + hex.EncodeToString(address.Bytes())
+	var blockNrOrHashValue interface{}
+	if blockNrOrHash != nil {
+		if blockNrOrHash.BlockNumber != nil {
+			bn := *blockNrOrHash.BlockNumber
+			blockNrOrHashValue = bn.MarshallJson()
+		} else if blockNrOrHash.BlockHash != nil {
+			blockNrOrHashValue = "0x" + hex.EncodeToString(blockNrOrHash.BlockHash.Bytes())
+		}
+	}
+
+	res, err := client.JSONRPCCall(rpcUrl, "eth_getTransactionCount", addressHex, blockNrOrHashValue)
+	if err != nil {
+		return nil, err
+	}
+
+	if res.Error != nil {
+		return nil, fmt.Errorf("RPC error response: %s", res.Error.Message)
+	}
+
+	//hash comes in escaped quotes, so we trim them here
+	// \"0x1234\" -> 0x1234
+	hashHex := strings.Trim(string(res.Result), "\"")
+
+	// now convert to a uint
+	decoded, err := hexutil.DecodeUint64(hashHex)
+	if err != nil {
+		return nil, err
+	}
+
+	result := hexutil.Uint64(decoded)
+
+	return &result, nil
+}

cmd/rpcdaemon/commands/txpool_api.go

@@ -14,30 +14,41 @@ import (
 	"github.com/ledgerwatch/erigon/core/rawdb"
 	"github.com/ledgerwatch/erigon/core/types"
 	"github.com/ledgerwatch/erigon/rlp"
+	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
 )
 
 // NetAPI the interface for the net_ RPC commands
 type TxPoolAPI interface {
-	Content(ctx context.Context) (map[string]map[string]map[string]*RPCTransaction, error)
+	Content(ctx context.Context) (interface{}, error)
 }
 
 // TxPoolAPIImpl data structure to store things needed for net_ commands
 type TxPoolAPIImpl struct {
 	*BaseAPI
-	pool proto_txpool.TxpoolClient
-	db   kv.RoDB
+	pool     proto_txpool.TxpoolClient
+	db       kv.RoDB
+	l2RPCUrl string
 }
 
 // NewTxPoolAPI returns NetAPIImplImpl instance
-func NewTxPoolAPI(base *BaseAPI, db kv.RoDB, pool proto_txpool.TxpoolClient) *TxPoolAPIImpl {
+func NewTxPoolAPI(base *BaseAPI, db kv.RoDB, pool proto_txpool.TxpoolClient, l2RPCUrl string) *TxPoolAPIImpl {
 	return &TxPoolAPIImpl{
-		BaseAPI: base,
-		pool:    pool,
-		db:      db,
+		BaseAPI:  base,
+		pool:     pool,
+		db:       db,
+		l2RPCUrl: l2RPCUrl,
 	}
 }
 
-func (api *TxPoolAPIImpl) Content(ctx context.Context) (map[string]map[string]map[string]*RPCTransaction, error) {
+func (api *TxPoolAPIImpl) Content(ctx context.Context) (interface{}, error) {
+	if api.l2RPCUrl != "" {
+		res, err := client.JSONRPCCall(api.l2RPCUrl, "txpool_content")
+		if err != nil {
+			return nil, err
+		}
+		return res.Result, nil
+	}
+
 	reply, err := api.pool.All(ctx, &proto_txpool.AllRequest{})
 	if err != nil {
 		return nil, err
@@ -120,7 +131,15 @@ func (api *TxPoolAPIImpl) Content(ctx context.Context) (map[string]map[string]ma
 }
 
 // Status returns the number of pending and queued transaction in the pool.
-func (api *TxPoolAPIImpl) Status(ctx context.Context) (map[string]hexutil.Uint, error) {
+func (api *TxPoolAPIImpl) Status(ctx context.Context) (interface{}, error) {
+	if api.l2RPCUrl != "" {
+		res, err := client.JSONRPCCall(api.l2RPCUrl, "txpool_status")
+		if err != nil {
+			return nil, err
+		}
+		return res.Result, nil
+	}
+
 	reply, err := api.pool.Status(ctx, &proto_txpool.StatusRequest{})
 	if err != nil {
 		return nil, err

rpc/types.go

@@ -131,6 +131,28 @@ func (bn *BlockNumber) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+func (bn *BlockNumber) MarshallJson() string {
+	if bn == nil {
+		return "latest"
+	}
+	switch *bn {
+	case 0:
+		return "earliest"
+	case -1:
+		return "latest"
+	case -2:
+		return "pending"
+	case -3:
+		return "safe"
+	case -4:
+		return "finalized"
+	case -5:
+		return "latestExecuted"
+	default:
+		return fmt.Sprintf("0x%x", *bn)
+	}
+}
+
 func (bn BlockNumber) Int64() int64 {
 	return int64(bn)
 }

zk/txpool/pool.go

@@ -1864,6 +1864,12 @@ func (b *BySenderAndNonce) nonce(senderID uint64) (nonce uint64, ok bool) {
 	s.Tx.Nonce = math.MaxUint64
 
 	b.tree.DescendLessOrEqual(s, func(mt *metaTx) bool {
+		if mt.currentSubPool != PendingSubPool {
+			// we only want to include transactions that are in the pending pool.  TXs in the queued pool
+			// artificially increase the "pending" call which can cause transactions to just stack up
+			// when libraries use eth_getTransactionCount "pending" for the next tx nonce - a common thing
+			return true
+		}
 		if mt.Tx.SenderID == senderID {
 			nonce = mt.Tx.Nonce
 			ok = true