Sequencing stage (#53)

* setting up some initial work to link sequences stage to the txpool

* vendoring the txpool into erigon

* transactions getting from the pool to the sequencer stage

* some progress on the sequencer creating a block

* interhashes progress for sequencer

* tweaks to sequencer work

* further work on sequencer

* txpool erigon-lib code removal

* some more txpool erigon-lib removal

* extracting some zk concerns from send raw transaction call

cmd/erigon-el/backend/backend.go

@@ -41,10 +41,10 @@ import (
 	"github.com/ledgerwatch/erigon-lib/kv/kvcfg"
 	"github.com/ledgerwatch/erigon-lib/kv/remotedbserver"
 	libstate "github.com/ledgerwatch/erigon-lib/state"
-	txpool2 "github.com/ledgerwatch/erigon-lib/txpool"
-	"github.com/ledgerwatch/erigon-lib/txpool/txpooluitl"
 	types2 "github.com/ledgerwatch/erigon-lib/types"
 	"github.com/ledgerwatch/erigon/chain"
+	txpool2 "github.com/ledgerwatch/erigon/zk/txpool"
+	"github.com/ledgerwatch/erigon/zk/txpool/txpooluitl"
 
 	"github.com/0xPolygonHermez/zkevm-data-streamer/datastreamer"
 	chain2 "github.com/ledgerwatch/erigon-lib/chain"

cmd/rpcdaemon/commands/send_transaction.go

@@ -5,22 +5,18 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"math/big"
-	"strings"
-
 	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/common/hexutility"
 	txPoolProto "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
 	"github.com/ledgerwatch/log/v3"
+	"math/big"
 
 	"github.com/ledgerwatch/erigon/core/rawdb"
 	"github.com/ledgerwatch/erigon/core/types"
 	"github.com/ledgerwatch/erigon/crypto"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/params"
 	"github.com/ledgerwatch/erigon/rlp"
-	"github.com/ledgerwatch/erigon/zk/zkchainconfig"
-	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
 )
 
 // SendRawTransaction implements eth_sendRawTransaction. Creates new message call transaction or a contract creation for previously-signed transactions.
@@ -36,8 +32,8 @@ func (api *APIImpl) SendRawTransaction(ctx context.Context, encodedTx hexutility
 	}
 	chainId := cc.ChainID
 
-	// [zkevm] - proxy the request if the chainID is ZK
-	if zkchainconfig.IsZk(chainId.Uint64()) {
+	// [zkevm] - proxy the request if the chainID is ZK and not a sequencer
+	if api.isZkNonSequencer(chainId) {
 		return api.sendTxZk(api.L2RpcUrl, encodedTx, chainId.Uint64())
 	}
 
@@ -92,24 +88,6 @@ func (api *APIImpl) SendRawTransaction(ctx context.Context, encodedTx hexutility
 	return txn.Hash(), nil
 }
 
-func (api *APIImpl) sendTxZk(rpcUrl string, encodedTx hexutility.Bytes, chainId uint64) (common.Hash, error) {
-
-	res, err := client.JSONRPCCall(rpcUrl, "eth_sendRawTransaction", encodedTx)
-	if err != nil {
-		return common.Hash{}, err
-	}
-
-	if res.Error != nil {
-		return common.Hash{}, 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), "\"")
-
-	return common.HexToHash(hashHex), nil
-}
-
 // SendTransaction implements eth_sendTransaction. Creates new message call transaction or a contract creation if the data field contains code.
 func (api *APIImpl) SendTransaction(_ context.Context, txObject interface{}) (common.Hash, error) {
 	return common.Hash{0}, fmt.Errorf(NotImplemented, "eth_sendTransaction")

cmd/rpcdaemon/commands/send_transaction_test.go

@@ -74,7 +74,7 @@ func TestSendRawTransaction(t *testing.T) {
 	ff := rpchelper.New(ctx, nil, txPool, txpool.NewMiningClient(conn), func() {})
 	stateCache := kvcache.New(kvcache.DefaultCoherentConfig)
 	br := snapshotsync.NewBlockReaderWithSnapshots(m.BlockSnapshots, m.TransactionsV3)
-	api := commands.NewEthAPI(commands.NewBaseApi(ff, stateCache, br, nil, false, rpccfg.DefaultEvmCallTimeout, m.Engine, m.Dirs), m.DB, nil, txPool, nil, 5000000, 100_000, "")
+	api := commands.NewEthAPI(commands.NewBaseApi(ff, stateCache, br, nil, false, rpccfg.DefaultEvmCallTimeout, m.Engine, m.Dirs, ""), m.DB, nil, txPool, nil, 5000000, 100_000, "")
 
 	buf := bytes.NewBuffer(nil)
 	err = txn.MarshalBinary(buf)

cmd/rpcdaemon/commands/send_transaction_zk.go

@@ -0,0 +1,34 @@
+package commands
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/ledgerwatch/erigon-lib/common"
+	"github.com/ledgerwatch/erigon-lib/common/hexutility"
+	"github.com/ledgerwatch/erigon/zk/sequencer"
+	"github.com/ledgerwatch/erigon/zk/zkchainconfig"
+	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
+	"math/big"
+)
+
+func (api *APIImpl) isZkNonSequencer(chainId *big.Int) bool {
+	return !sequencer.IsSequencer() && zkchainconfig.IsZk(chainId.Uint64())
+}
+
+func (api *APIImpl) sendTxZk(rpcUrl string, encodedTx hexutility.Bytes, chainId uint64) (common.Hash, error) {
+	res, err := client.JSONRPCCall(rpcUrl, "eth_sendRawTransaction", encodedTx)
+	if err != nil {
+		return common.Hash{}, err
+	}
+
+	if res.Error != nil {
+		return common.Hash{}, 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), "\"")
+
+	return common.HexToHash(hashHex), nil
+}

cmd/rpcdaemon/rpcservices/eth_txpool.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ledgerwatch/erigon-lib/gointerfaces"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/grpcutil"
 	txpooproto "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
-	txpool2 "github.com/ledgerwatch/erigon-lib/txpool"
+	txpool2 "github.com/ledgerwatch/erigon/zk/txpool"
 	"github.com/ledgerwatch/log/v3"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/emptypb"

cmd/txpool/main.go

@@ -18,13 +18,13 @@ import (
 	"github.com/ledgerwatch/erigon-lib/kv/kvcache"
 	"github.com/ledgerwatch/erigon-lib/kv/remotedb"
 	"github.com/ledgerwatch/erigon-lib/kv/remotedbserver"
-	"github.com/ledgerwatch/erigon-lib/txpool"
 	"github.com/ledgerwatch/erigon-lib/txpool/txpoolcfg"
-	"github.com/ledgerwatch/erigon-lib/txpool/txpooluitl"
 	"github.com/ledgerwatch/erigon-lib/types"
 	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/rpcdaemontest"
 	common2 "github.com/ledgerwatch/erigon/common"
 	"github.com/ledgerwatch/erigon/ethdb/privateapi"
+	"github.com/ledgerwatch/erigon/zk/txpool"
+	"github.com/ledgerwatch/erigon/zk/txpool/txpooluitl"
 	"github.com/ledgerwatch/log/v3"
 	"github.com/spf13/cobra"
 

core/state_transition.go

@@ -22,8 +22,8 @@ import (
 	"github.com/holiman/uint256"
 
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
-	"github.com/ledgerwatch/erigon-lib/txpool"
 	types2 "github.com/ledgerwatch/erigon-lib/types"
+	"github.com/ledgerwatch/erigon/zk/txpool"
 
 	"github.com/ledgerwatch/erigon/common"
 	cmath "github.com/ledgerwatch/erigon/common/math"

eth/backend.go

@@ -58,9 +58,8 @@ import (
 	"github.com/ledgerwatch/erigon-lib/kv/kvcfg"
 	"github.com/ledgerwatch/erigon-lib/kv/remotedbserver"
 	libstate "github.com/ledgerwatch/erigon-lib/state"
-	txpool2 "github.com/ledgerwatch/erigon-lib/txpool"
-	"github.com/ledgerwatch/erigon-lib/txpool/txpooluitl"
 	types2 "github.com/ledgerwatch/erigon-lib/types"
+	"github.com/ledgerwatch/erigon/zk/txpool/txpooluitl"
 
 	"github.com/ledgerwatch/erigon/chain"
 
@@ -115,6 +114,7 @@ import (
 	"github.com/ledgerwatch/erigon/zk/hermez_db"
 	zkStages "github.com/ledgerwatch/erigon/zk/stages"
 	"github.com/ledgerwatch/erigon/zk/syncer"
+	txpool2 "github.com/ledgerwatch/erigon/zk/txpool"
 	"github.com/ledgerwatch/erigon/zkevm/etherman"
 )
 
@@ -733,6 +733,8 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 				backend.forkValidator,
 				backend.engine,
 				backend.dataStream,
+				backend.txPool2,
+				backend.txPool2DB,
 			)
 
 			backend.syncUnwindOrder = zkStages.ZkSequencerUnwindOrder

eth/stagedsync/stage_mining_create_block.go

@@ -12,7 +12,7 @@ import (
 	"github.com/ledgerwatch/log/v3"
 
 	"github.com/ledgerwatch/erigon-lib/kv"
-	"github.com/ledgerwatch/erigon-lib/txpool"
+	"github.com/ledgerwatch/erigon/zk/txpool"
 
 	"github.com/ledgerwatch/erigon/common/debug"
 	"github.com/ledgerwatch/erigon/consensus"

eth/stagedsync/stage_mining_exec.go

@@ -20,8 +20,8 @@ import (
 
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon-lib/kv/memdb"
-	"github.com/ledgerwatch/erigon-lib/txpool"
 	types2 "github.com/ledgerwatch/erigon-lib/types"
+	"github.com/ledgerwatch/erigon/zk/txpool"
 
 	"github.com/ledgerwatch/erigon/core/types/accounts"
 	"github.com/ledgerwatch/erigon/rlp"

turbo/rpchelper/filters.go

@@ -17,7 +17,7 @@ import (
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/grpcutil"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/remote"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
-	txpool2 "github.com/ledgerwatch/erigon-lib/txpool"
+	txpool2 "github.com/ledgerwatch/erigon/zk/txpool"
 	"github.com/ledgerwatch/log/v3"
 	"google.golang.org/grpc"
 

turbo/stages/mock_sentry.go

@@ -26,10 +26,10 @@ import (
 	"github.com/ledgerwatch/erigon-lib/kv/memdb"
 	"github.com/ledgerwatch/erigon-lib/kv/remotedbserver"
 	libstate "github.com/ledgerwatch/erigon-lib/state"
-	"github.com/ledgerwatch/erigon-lib/txpool"
 	"github.com/ledgerwatch/erigon-lib/txpool/txpoolcfg"
 	types2 "github.com/ledgerwatch/erigon-lib/types"
 	"github.com/ledgerwatch/erigon/chain"
+	"github.com/ledgerwatch/erigon/zk/txpool"
 	"github.com/ledgerwatch/log/v3"
 	"google.golang.org/protobuf/types/known/emptypb"
 
@@ -307,7 +307,6 @@ func MockWithEverything(t *testing.T, gspec *types.Genesis, key *ecdsa.PrivateKe
 	blockPropagator := func(Ctx context.Context, header *types.Header, body *types.RawBody, td *big.Int) {}
 
 	if !cfg.DeprecatedTxPool.Disable {
-		poolCfg := txpoolcfg.DefaultConfig
 		newTxs := make(chan types2.Announcements, 1024)
 		if t != nil {
 			t.Cleanup(func() {
@@ -316,7 +315,7 @@ func MockWithEverything(t *testing.T, gspec *types.Genesis, key *ecdsa.PrivateKe
 		}
 		chainID, _ := uint256.FromBig(mock.ChainConfig.ChainID)
 		shanghaiTime := mock.ChainConfig.ShanghaiTime
-		mock.TxPool, err = txpool.New(newTxs, mock.DB, poolCfg, kvcache.NewDummy(), *chainID, shanghaiTime)
+		mock.TxPool, err = txpool.New(newTxs, mock.DB, txpoolcfg.DefaultConfig, kvcache.NewDummy(), *chainID, shanghaiTime)
 		if err != nil {
 			t.Fatal(err)
 		}

turbo/stages/zk_stages.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ledgerwatch/erigon/zk/datastream/client"
 	zkStages "github.com/ledgerwatch/erigon/zk/stages"
 	"github.com/ledgerwatch/erigon/zk/syncer"
+	"github.com/ledgerwatch/erigon/zk/txpool"
 )
 
 // NewDefaultZkStages creates stages for zk syncer (RPC mode)
@@ -92,6 +93,8 @@ func NewSequencerZkStages(ctx context.Context,
 	forkValidator *engineapi.ForkValidator,
 	engine consensus.Engine,
 	datastreamServer *datastreamer.StreamServer,
+	txPool *txpool.TxPool,
+	txPoolDb kv.RwDB,
 ) []*stagedsync.Stage {
 	dirs := cfg.Dirs
 	blockReader := snapshotsync.NewBlockReaderWithSnapshots(snapshots, cfg.TransactionsV3)
@@ -103,6 +106,7 @@ func NewSequencerZkStages(ctx context.Context,
 	return zkStages.SequencerZkStages(ctx,
 		stagedsync.StageCumulativeIndexCfg(db),
 		zkStages.StageDataStreamCatchupCfg(datastreamServer, db),
+		zkStages.StageSequencerInterhashesCfg(db),
 		zkStages.StageSequenceBlocksCfg(
 			db,
 			cfg.Prune,
@@ -117,11 +121,12 @@ func NewSequencerZkStages(ctx context.Context,
 			cfg.HistoryV3,
 			dirs,
 			blockReader,
-			controlServer.Hd,
 			cfg.Genesis,
 			cfg.Sync,
 			agg,
 			cfg.Zk,
+			txPool,
+			txPoolDb,
 		),
 		stagedsync.StageHashStateCfg(db, dirs, cfg.HistoryV3, agg),
 		zkStages.StageZkInterHashesCfg(db, true, true, false, dirs.Tmp, blockReader, controlServer.Hd, cfg.HistoryV3, agg, cfg.Zk),

zk/stages/stage_interhashes.go

@@ -30,7 +30,7 @@ import (
 	"github.com/ledgerwatch/erigon/core/types/accounts"
 	"github.com/ledgerwatch/erigon/eth/ethconfig"
 	"github.com/ledgerwatch/erigon/eth/stagedsync"
-	stages "github.com/ledgerwatch/erigon/eth/stagedsync/stages"
+	"github.com/ledgerwatch/erigon/eth/stagedsync/stages"
 	"github.com/ledgerwatch/erigon/turbo/services"
 	"github.com/ledgerwatch/erigon/turbo/stages/headerdownload"
 	"github.com/ledgerwatch/erigon/turbo/trie"