(feat): preconf + private tx relay #2004

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

manav2401 wants to merge 39 commits into develop from preconf-private-tx-relay

eth/api_backend.go

-Original file line number
+Diff line change
@@ Expand Up / @@ -39,6 +39,7 @@ import ( @@
     	"github.com/ethereum/go-ethereum/core/types"
     	"github.com/ethereum/go-ethereum/core/vm"
     	"github.com/ethereum/go-ethereum/eth/gasprice"
+    	"github.com/ethereum/go-ethereum/eth/relay"
     	"github.com/ethereum/go-ethereum/eth/tracers"
     	"github.com/ethereum/go-ethereum/ethdb"
     	"github.com/ethereum/go-ethereum/event"
@@ Expand All / @@ -54,6 +55,8 @@ type EthAPIBackend struct { @@
     	allowUnprotectedTxs bool
     	eth                 *Ethereum
     	gpo                 *gasprice.Oracle
+    	relay *relay.RelayService
     }
     // ChainConfig returns the active chain configuration.
@@ Expand Down Expand Up @@
     	return b.eth.txPool.ContentFrom(addr)
     }
+    func (b *EthAPIBackend) TxStatus(hash common.Hash) txpool.TxStatus {
+    	return b.eth.txPool.Status(hash)
+    }
     func (b *EthAPIBackend) TxPool() *txpool.TxPool {
     	return b.eth.txPool
     }
@@ Expand Down Expand Up @@
     func (b *EthAPIBackend) RPCTxSyncMaxTimeout() time.Duration {
     	return b.eth.config.TxSyncMaxTimeout
     }
+    // Preconf / Private tx related API for relay
+    func (b *EthAPIBackend) PreconfEnabled() bool {
+    	return b.relay.PreconfEnabled()
+    }
+    func (b *EthAPIBackend) SubmitTxForPreconf(tx *types.Transaction) error {
+    	return b.relay.SubmitPreconfTransaction(tx)
+    }
+    func (b *EthAPIBackend) CheckPreconfStatus(hash common.Hash) (bool, error) {
+    	return b.relay.CheckPreconfStatus(hash)
+    }
+    func (b *EthAPIBackend) PrivateTxEnabled() bool {
+    	return b.relay.PrivateTxEnabled()
+    }
+    func (b *EthAPIBackend) SubmitPrivateTx(tx *types.Transaction) error {
+    	return b.relay.SubmitPrivateTransaction(tx)
+    }
+    // Preconf / Private tx related API for block producers
+    func (b *EthAPIBackend) AcceptPreconfTxs() bool {
+    	return b.relay.AcceptPreconfTxs()
+    }
+    func (b *EthAPIBackend) AcceptPrivateTxs() bool {
+    	return b.relay.AcceptPrivateTxs()
+    }
+    func (b *EthAPIBackend) RecordPrivateTx(hash common.Hash) {
+    	b.relay.RecordPrivateTx(hash)
+    }
+    func (b *EthAPIBackend) PurgePrivateTx(hash common.Hash) {
+    	b.relay.PurgePrivateTx(hash)
+    }

eth/backend.go

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -54,6 +54,7 @@ import (
  
    	"github.com/ethereum/go-ethereum/eth/protocols/eth"

    	"github.com/ethereum/go-ethereum/eth/protocols/snap"

    	"github.com/ethereum/go-ethereum/eth/protocols/wit"

    	"github.com/ethereum/go-ethereum/eth/relay"

    	"github.com/ethereum/go-ethereum/eth/tracers"

    	"github.com/ethereum/go-ethereum/ethdb"

    	"github.com/ethereum/go-ethereum/event"

    @@ -228,14 +229,18 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
  
    		closeCh:         make(chan struct{}),

    	}

    	relayService := relay.Init(config.EnablePreconfs, config.EnablePrivateTx, config.AcceptPreconfTx, config.AcceptPrivateTx, config.BlockProducerRpcEndpoints)

    	privateTxGetter := relayService.GetPrivateTxGetter()

    	// START: Bor changes

    	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, eth, nil}

    	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, eth, nil, relayService}

    	if eth.APIBackend.allowUnprotectedTxs {

    		log.Info("------Unprotected transactions allowed-------")

    		log.Info("Unprotected transactions allowed")

    		config.TxPool.AllowUnprotectedTxs = true

    	}

    	gpoParams := config.GPO

    	// Set transaction getter for relay service to query local database

    	relayService.SetTxGetter(eth.APIBackend.GetCanonicalTransaction)

    	blockChainAPI := ethapi.NewBlockChainAPI(eth.APIBackend)

    	engine, err := ethconfig.CreateConsensusEngine(config.Genesis.Config, config, chainDb, blockChainAPI)

    @@ -327,6 +332,9 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
  
    		eth.blockchain, err = core.NewBlockChain(chainDb, config.Genesis, eth.engine, options)

    	}

    	// Set the chain head event subscription function for private tx store

    	relayService.SetchainEventSubFn(eth.blockchain.SubscribeChainEvent)

    	// Set parallel stateless import toggle on blockchain

    	if err == nil && eth.blockchain != nil && config.EnableParallelStatelessImport {

    		eth.blockchain.ParallelStatelessImportEnable()

    @@ -342,18 +350,6 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
  
    	// Set blockchain reference for fork detection in whitelist service

    	checker.SetBlockchain(eth.blockchain)

    	// 1.14.8: NewOracle function definition was changed to accept (startPrice *big.Int) param.

    	eth.APIBackend.gpo = gasprice.NewOracle(eth.APIBackend, gpoParams, config.Miner.GasPrice)

    	// bor: this is nor present in geth

    	/*

    		_ = eth.engine.VerifyHeader(eth.blockchain, eth.blockchain.CurrentHeader()) // TODO think on it

    	*/

    	// BOR changes

    	eth.APIBackend.gpo.ProcessCache()

    	// BOR changes

    	// Initialize filtermaps log index.

    	fmConfig := filtermaps.Config{

    		History:        config.LogHistory,

    @@ -423,11 +419,13 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
  
    		checker:                 checker,

    		enableBlockTracking:     eth.config.EnableBlockTracking,

    		txAnnouncementOnly:      eth.p2pServer.TxAnnouncementOnly,

    		disableTxPropagation:    eth.p2pServer.DisableTxPropagation,

    		witnessProtocol:         eth.config.WitnessProtocol,

    		syncWithWitnesses:       eth.config.SyncWithWitnesses,

    		syncAndProduceWitnesses: eth.config.SyncAndProduceWitnesses,

    		fastForwardThreshold:    config.FastForwardThreshold,

    		p2pServer:               eth.p2pServer,

    		privateTxGetter:         privateTxGetter,

    	}); err != nil {

    		return nil, err

    	}

    @@ -440,12 +438,9 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
  
    		eth.miner.SetPrioAddresses(config.TxPool.Locals)

    	}

    	eth.APIBackend = &EthAPIBackend{stack.Config().ExtRPCEnabled(), stack.Config().AllowUnprotectedTxs, eth, nil}

    	if eth.APIBackend.allowUnprotectedTxs {

    		log.Info("Unprotected transactions allowed")

    	}

    	// 1.14.8: NewOracle function definition was changed to accept (startPrice *big.Int) param.

    	eth.APIBackend.gpo = gasprice.NewOracle(eth.APIBackend, config.GPO, config.Miner.GasPrice)

    	eth.APIBackend.gpo.ProcessCache()

    	// Start the RPC service

    	eth.netRPCService = ethapi.NewNetAPI(eth.p2pServer, config.NetworkId)

    @@ -1010,6 +1005,11 @@ func (s *Ethereum) Stop() error {
  
    	// Stop all the peer-related stuff first.

    	s.discmix.Close()

    	// Close the tx relay service if enabled

    	if s.APIBackend.relay != nil {

    		s.APIBackend.relay.Close()

    	}

    	// Close the engine before handler else it may cause a deadlock where

    	// the heimdall is unresponsive and the syncing loop keeps waiting

    	// for a response and is unable to proceed to exit `Finalize` during

eth/ethconfig/config.go

-Original file line number
+Diff line change
@@ Expand Up / @@ -288,6 +288,15 @@ type Config struct { @@
     	// EIP-7966: eth_sendRawTransactionSync timeouts
     	TxSyncDefaultTimeout time.Duration `toml:",omitempty"`
     	TxSyncMaxTimeout     time.Duration `toml:",omitempty"`
+    	// Preconf / Private transaction relay related settings
+    	EnablePreconfs            bool
+    	EnablePrivateTx           bool
+    	BlockProducerRpcEndpoints []string
+    	// Preconf / Private transaction related settings for block producers
+    	AcceptPreconfTx bool
+    	AcceptPrivateTx bool
     }
     // CreateConsensusEngine creates a consensus engine for the given chain configuration.
@@ Expand Down @@

eth/handler.go

-Original file line number
+Diff line change
@@ Expand Up / @@ -44,6 +44,7 @@ import ( @@
     	"github.com/ethereum/go-ethereum/eth/protocols/eth"
     	"github.com/ethereum/go-ethereum/eth/protocols/snap"
     	"github.com/ethereum/go-ethereum/eth/protocols/wit"
+    	"github.com/ethereum/go-ethereum/eth/relay"
     	"github.com/ethereum/go-ethereum/ethdb"
     	"github.com/ethereum/go-ethereum/event"
     	"github.com/ethereum/go-ethereum/internal/ethapi"
@@ Expand Down Expand Up / @@ -120,12 +121,14 @@ type handlerConfig struct { @@
     	EthAPI                  *ethapi.BlockChainAPI  // EthAPI to interact
     	enableBlockTracking     bool                   // Whether to log information collected while tracking block lifecycle
     	txAnnouncementOnly      bool                   // Whether to only announce txs to peers
+    	disableTxPropagation    bool                   // Whether to disable broadcasting and announcement of txs to peers
     	witnessProtocol         bool                   // Whether to enable witness protocol
     	syncWithWitnesses       bool                   // Whether to sync blocks with witnesses
     	syncAndProduceWitnesses bool                   // Whether to sync blocks and produce witnesses simultaneously
     	fastForwardThreshold    uint64                 // Minimum necessary distance between local header and peer to fast forward
     	gasCeil                 uint64                 // Gas ceiling for dynamic witness page threshold calculation
     	p2pServer               *p2p.Server            // P2P server for jailing peers
+    	privateTxGetter         relay.PrivateTxGetter  // privateTxGetter to check if a transaction needs to be treated as private or not
     }
     type handler struct {
@@ Expand All / @@ -150,6 +153,9 @@ type handler struct { @@
     	ethAPI *ethapi.BlockChainAPI // EthAPI to interact
+    	// privateTxGetter to check if a transaction needs to be treated as private or not
+    	privateTxGetter relay.PrivateTxGetter
     	eventMux      *event.TypeMux
     	txsCh         chan core.NewTxsEvent
     	txsSub        event.Subscription
@@ Expand All / @@ -160,9 +166,11 @@ type handler struct { @@
     	requiredBlocks map[uint64]common.Hash
-    	enableBlockTracking bool
-    	txAnnouncementOnly  bool
-    	p2pServer           *p2p.Server // P2P server for jailing peers
+    	enableBlockTracking  bool
+    	txAnnouncementOnly   bool
+    	disableTxPropagation bool
+    	p2pServer *p2p.Server // P2P server for jailing peers
     	// Witness protocol related fields
     	syncWithWitnesses       bool
@@ Expand Down Expand Up @@
     		requiredBlocks:          config.RequiredBlocks,
     		enableBlockTracking:     config.enableBlockTracking,
     		txAnnouncementOnly:      config.txAnnouncementOnly,
+    		disableTxPropagation:    config.disableTxPropagation,
     		p2pServer:               config.p2pServer,
     		quitSync:                make(chan struct{}),
     		handlerDoneCh:           make(chan struct{}),
     		handlerStartCh:          make(chan struct{}),
     		syncWithWitnesses:       config.syncWithWitnesses,
     		syncAndProduceWitnesses: config.syncAndProduceWitnesses,
+    		privateTxGetter:         config.privateTxGetter,
     	}
     	log.Info("Sync with witnesses", "enabled", config.syncWithWitnesses)
@@ Expand Down Expand Up @@
     	}
     	h.chainSync.handlePeerEvent()
-    	// Propagate existing transactions. new transactions appearing
-    	// after this will be sent via broadcasts.
-    	h.syncTransactions(peer)
+    	// Bor: skip propagating transactions if flag is set
+    	if !h.disableTxPropagation {
+    		// Propagate existing transactions. new transactions appearing
+    		// after this will be sent via broadcasts.
+    		h.syncTransactions(peer)
+    	}
     	// Create a notification channel for pending requests if the peer goes down
     	dead := make(chan struct{})
@@ Expand Down Expand Up / @@ -579,17 +592,27 @@ func (h *handler) unregisterPeer(id string) { @@
     func (h *handler) Start(maxPeers int) {
     	h.maxPeers = maxPeers
-    	// broadcast and announce transactions (only new ones, not resurrected ones)
-    	h.wg.Add(1)
-    	h.txsCh = make(chan core.NewTxsEvent, txChanSize)
-    	h.txsSub = h.txpool.SubscribeTransactions(h.txsCh, false)
-    	go h.txBroadcastLoop()
+    	if h.disableTxPropagation {
+    		log.Info("Disabling transaction propagation completely")
+    	}
+    	// Bor: block producers can choose to not propagate transactions to save p2p overhead
+    	// broadcast and announce transactions (only new ones, not resurrected ones) only
+    	// if transaction propagation is enabled
+    	if !h.disableTxPropagation {
+    		h.wg.Add(1)
+    		h.txsCh = make(chan core.NewTxsEvent, txChanSize)
+    		h.txsSub = h.txpool.SubscribeTransactions(h.txsCh, false)
+    		go h.txBroadcastLoop()
+    	}
     	// rebroadcast stuck transactions
-    	h.wg.Add(1)
-    	h.stuckTxsCh = make(chan core.StuckTxsEvent, txChanSize)
-    	h.stuckTxsSub = h.txpool.SubscribeRebroadcastTransactions(h.stuckTxsCh)
-    	go h.stuckTxBroadcastLoop()
+    	if !h.disableTxPropagation {
+    		h.wg.Add(1)
+    		h.stuckTxsCh = make(chan core.StuckTxsEvent, txChanSize)
+    		h.stuckTxsSub = h.txpool.SubscribeRebroadcastTransactions(h.stuckTxsCh)
+    		go h.stuckTxBroadcastLoop()
+    	}
     	// broadcast mined blocks
     	h.wg.Add(1)
@@ Expand All / @@ -610,8 +633,12 @@ func (h *handler) Start(maxPeers int) { @@
     }
     func (h *handler) Stop() {
-    	h.txsSub.Unsubscribe()      // quits txBroadcastLoop
-    	h.stuckTxsSub.Unsubscribe() // quits stuckTxBroadcastLoop
+    	if h.txsSub != nil {
+    		h.txsSub.Unsubscribe() // quits txBroadcastLoop
+    	}
+    	if h.stuckTxsSub != nil {
+    		h.stuckTxsSub.Unsubscribe() // quits stuckTxBroadcastLoop
+    	}
     	h.minedBlockSub.Unsubscribe()
     	h.blockRange.stop()
@@ Expand Down Expand Up @@
     	)
     	for _, tx := range txs {
+    		// Skip gossip if transaction is marked as private
+    		if h.privateTxGetter != nil && h.privateTxGetter.IsTxPrivate(tx.Hash()) {
+    			log.Debug("[tx-relay] skip tx broadcast for private tx", "hash", tx.Hash())
+    			continue
+    		}
     		var directSet map[*ethPeer]struct{}
     		switch {
     		case tx.Type() == types.BlobTxType:
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

(feat): preconf + private tx relay #2004

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

(feat): preconf + private tx relay #2004

Are you sure you want to change the base?

Uh oh!

(feat): preconf + private tx relay #2004

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!