feat: forward txs to sequencer (#1208)

* add flags to disable broadcast and receiving of tx gossip respectively

* feat: support forward txs to sequencer

* fix: config

* chore: auto version bump [bot]

* chore: auto version bump [bot]

* add comments

* fix: forward tx logic

* chore: auto version bump [bot]

* fix: comment typo

* chore: auto version bump [bot]

* chore: auto version bump [bot]

* run goimport

* fix another file

* fix: comments

* chore: auto version bump [bot]

* feat: retain transaction before forwording

* chore: auto version bump [bot]

---------

Co-authored-by: jonastheis <[email protected]>
Co-authored-by: colin <[email protected]>
Co-authored-by: colinlyguo <[email protected]>

Author: 4 people

cmd/geth/main.go

@@ -178,6 +178,7 @@ var (
 		utils.ShadowforkPeersFlag,
 		utils.TxGossipBroadcastDisabledFlag,
 		utils.TxGossipReceivingDisabledFlag,
+		utils.TxGossipSequencerHTTPFlag,
 		utils.DASyncEnabledFlag,
 		utils.DAMissingHeaderFieldsBaseURLFlag,
 		utils.DABlockNativeAPIEndpointFlag,

cmd/geth/usage.go

@@ -252,6 +252,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.CircuitCapacityCheckWorkersFlag,
 			utils.TxGossipBroadcastDisabledFlag,
 			utils.TxGossipReceivingDisabledFlag,
+			utils.TxGossipSequencerHTTPFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -895,13 +895,17 @@ var (
 
 	// Tx gossip settings
 	TxGossipBroadcastDisabledFlag = cli.BoolFlag{
-		Name:  "txgossip.disablebroadcast",
+		Name:  "gossip.disablebroadcast",
 		Usage: "Disable gossip broadcast transactions to other peers",
 	}
 	TxGossipReceivingDisabledFlag = cli.BoolFlag{
-		Name:  "txgossip.disablereceiving",
+		Name:  "gossip.disablereceiving",
 		Usage: "Disable gossip receiving transactions from other peers",
 	}
+	TxGossipSequencerHTTPFlag = &cli.StringFlag{
+		Name:  "gossip.sequencerhttp",
+		Usage: "Sequencer mempool HTTP endpoint",
+	}
 
 	// DA syncing settings
 	DASyncEnabledFlag = cli.BoolFlag{
@@ -1823,6 +1827,10 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
 		cfg.TxGossipReceivingDisabled = ctx.GlobalBool(TxGossipReceivingDisabledFlag.Name)
 		log.Info("Transaction gossip receiving disabled", "disabled", cfg.TxGossipReceivingDisabled)
 	}
+	// Only configure sequencer http flag if we're running in verifier mode i.e. --mine is disabled.
+	if ctx.IsSet(TxGossipSequencerHTTPFlag.Name) && !ctx.IsSet(MiningEnabledFlag.Name) {
+		cfg.TxGossipSequencerHTTP = ctx.String(TxGossipSequencerHTTPFlag.Name)
+	}
 
 	// Cap the cache allowance and tune the garbage collector
 	mem, err := gopsutil.VirtualMemory()

eth/api_backend.go

@@ -25,6 +25,7 @@ import (
 	"github.com/scroll-tech/go-ethereum"
 	"github.com/scroll-tech/go-ethereum/accounts"
 	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/common/hexutil"
 	"github.com/scroll-tech/go-ethereum/consensus"
 	"github.com/scroll-tech/go-ethereum/core"
 	"github.com/scroll-tech/go-ethereum/core/bloombits"
@@ -35,6 +36,7 @@ import (
 	"github.com/scroll-tech/go-ethereum/eth/gasprice"
 	"github.com/scroll-tech/go-ethereum/ethdb"
 	"github.com/scroll-tech/go-ethereum/event"
+	"github.com/scroll-tech/go-ethereum/log"
 	"github.com/scroll-tech/go-ethereum/miner"
 	"github.com/scroll-tech/go-ethereum/params"
 	"github.com/scroll-tech/go-ethereum/rpc"
@@ -44,6 +46,7 @@ import (
 type EthAPIBackend struct {
 	extRPCEnabled       bool
 	allowUnprotectedTxs bool
+	disableTxPool       bool
 	eth                 *Ethereum
 	gpo                 *gasprice.Oracle
 }
@@ -262,6 +265,34 @@ func (b *EthAPIBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscri
 }
 
 func (b *EthAPIBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error {
+	if signedTx.Type() == types.BlobTxType {
+		return types.ErrTxTypeNotSupported
+	}
+
+	// Retain tx in local tx pool before forwarding to sequencer rpc, for local RPC usage.
+	err := b.sendTx(signedTx)
+	if err != nil {
+		return err
+	}
+
+	// Forward to remote sequencer RPC
+	if b.eth.sequencerRPCService != nil {
+		signedTxData, err := signedTx.MarshalBinary()
+		if err != nil {
+			return err
+		}
+		if err = b.eth.sequencerRPCService.CallContext(ctx, nil, "eth_sendRawTransaction", hexutil.Encode(signedTxData)); err != nil {
+			log.Warn("failed to forward tx to sequencer", "tx", signedTx.Hash(), "err", err)
+			if b.disableTxPool {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (b *EthAPIBackend) sendTx(signedTx *types.Transaction) error {
 	// will `VerifyFee` & `validateTx` in txPool.AddLocal
 	return b.eth.txPool.AddLocal(signedTx)
 }

eth/backend.go

@@ -113,6 +113,9 @@ type Ethereum struct {
 	p2pServer *p2p.Server
 
 	lock sync.RWMutex // Protects the variadic fields (e.g. gas price and etherbase)
+
+	// Scroll additions
+	sequencerRPCService *rpc.Client
 }
 
 // New creates a new Ethereum object (including the
@@ -303,7 +306,7 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 	// Some of the extraData is used with Clique consensus (before EuclidV2). After EuclidV2 we use SystemContract consensus where this is overridden when creating a block.
 	eth.miner.SetExtra(makeExtraData(config.Miner.ExtraData))
 
-	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, eth, nil}
+	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, config.TxGossipReceivingDisabled, eth, nil}
 	if eth.APIBackend.allowUnprotectedTxs {
 		log.Info("Unprotected transactions allowed")
 	}
@@ -314,6 +317,16 @@ func New(stack *node.Node, config *ethconfig.Config, l1Client l1.Client) (*Ether
 	gpoParams.DefaultBasePrice = new(big.Int).SetUint64(config.TxPool.PriceLimit)
 	eth.APIBackend.gpo = gasprice.NewOracle(eth.APIBackend, gpoParams)
 
+	if config.TxGossipSequencerHTTP != "" {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		client, err := rpc.DialContext(ctx, config.TxGossipSequencerHTTP)
+		cancel()
+		if err != nil {
+			return nil, fmt.Errorf("cannot initialize rollup sequencer client: %w", err)
+		}
+		eth.sequencerRPCService = client
+	}
+
 	// Setup DNS discovery iterators.
 	dnsclient := dnsdisc.NewClient(dnsdisc.Config{})
 	eth.ethDialCandidates, err = dnsclient.NewIterator(eth.config.EthDiscoveryURLs...)
@@ -702,6 +715,9 @@ func (s *Ethereum) Stop() error {
 	}
 	s.blockchain.Stop()
 	s.engine.Close()
+	if s.sequencerRPCService != nil {
+		s.sequencerRPCService.Close()
+	}
 	rawdb.PopUncleanShutdownMarker(s.chainDb)
 	s.chainDb.Close()
 	s.eventMux.Stop()

eth/ethconfig/config.go

@@ -233,6 +233,7 @@ type Config struct {
 
 	TxGossipBroadcastDisabled bool
 	TxGossipReceivingDisabled bool
+	TxGossipSequencerHTTP     string
 }
 
 // CreateConsensusEngine creates a consensus engine for the given chain configuration.

params/version.go

@@ -24,7 +24,7 @@ import (
 const (
 	VersionMajor = 5         // Major version component of the current release
 	VersionMinor = 8         // Minor version component of the current release
-	VersionPatch = 66        // Patch version component of the current release
+	VersionPatch = 67        // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )