upstream: code review fix (ethereum#31)

* rerun go generate

* fix log in worker

* enable blobpool

* add SubscribeReannoTxsEvent for blobpool

* eth: fix potential hang in waitXXXExtension

* fix networkID when new eth handler

* remove unexpected +

* disable SyncTargetFlag

* add commented code in enableSyncedFeatures

* fix typo

cmd/devp2p/internal/ethtest/conn.go

@@ -342,7 +342,7 @@ loop:
 			if err := c.Write(ethProto, eth.StatusMsg, status); err != nil {
 				return fmt.Errorf("write to connection failed: %v", err)
 			}
-		case eth.UpgradeStatusMsg + +protoOffset(ethProto):
+		case eth.UpgradeStatusMsg + protoOffset(ethProto):
 			msg := new(eth.UpgradeStatusPacket)
 			if err := rlp.DecodeBytes(data, &msg); err != nil {
 				return fmt.Errorf("error decoding status packet: %w", err)

cmd/geth/main.go

@@ -93,7 +93,7 @@ var (
 		utils.BlobPoolPriceBumpFlag,
 		utils.SyncModeFlag,
 		utils.TriesVerifyModeFlag,
-		utils.SyncTargetFlag,
+		// utils.SyncTargetFlag,
 		utils.ExitWhenSyncedFlag,
 		utils.GCModeFlag,
 		utils.SnapshotFlag,

core/genesis.go

@@ -64,9 +64,9 @@ type Genesis struct {
 	Number        uint64      `json:"number"`
 	GasUsed       uint64      `json:"gasUsed"`
 	ParentHash    common.Hash `json:"parentHash"`
-	BaseFee       *big.Int    `json:"baseFeePerGas"`           // EIP-1559
-	ExcessBlobGas *uint64     `json:"excessBlobGas,omitempty"` // EIP-4844, TODO(Nathan): remove tag `omitempty` after cancun fork
-	BlobGasUsed   *uint64     `json:"blobGasUsed,omitempty"`   // EIP-4844, TODO(Nathan): remove tag `omitempty` after cancun fork
+	BaseFee       *big.Int    `json:"baseFeePerGas"` // EIP-1559
+	ExcessBlobGas *uint64     `json:"excessBlobGas"` // EIP-4844
+	BlobGasUsed   *uint64     `json:"blobGasUsed"`   // EIP-4844
 }
 
 func ReadGenesis(db ethdb.Database) (*Genesis, error) {

core/txpool/blobpool/blobpool.go

@@ -313,6 +313,8 @@ type BlobPool struct {
 
 	discoverFeed event.Feed // Event feed to send out new tx events on pool discovery (reorg excluded)
 	insertFeed   event.Feed // Event feed to send out new tx events on pool inclusion (reorg included)
+	reannoTxFeed event.Feed // Event feed for announcing transactions again
+	scope        event.SubscriptionScope
 
 	lock sync.RWMutex // Mutex protecting the pool during reorg handling
 }
@@ -1502,7 +1504,7 @@ func (p *BlobPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool
 // SubscribeReannoTxsEvent registers a subscription of ReannoTxsEvent and
 // starts sending event to the given channel.
 func (pool *BlobPool) SubscribeReannoTxsEvent(ch chan<- core.ReannoTxsEvent) event.Subscription {
-	panic("not supported")
+	return pool.scope.Track(pool.reannoTxFeed.Subscribe(ch))
 }
 
 // Nonce returns the next nonce of an account, with all transactions executable

eth/backend.go

@@ -37,6 +37,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state/pruner"
 	"github.com/ethereum/go-ethereum/core/txpool"
+	"github.com/ethereum/go-ethereum/core/txpool/blobpool"
 	"github.com/ethereum/go-ethereum/core/txpool/legacypool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
@@ -282,16 +283,14 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 	if config.BlobPool.Datadir != "" {
 		config.BlobPool.Datadir = stack.ResolvePath(config.BlobPool.Datadir)
 	}
-	// TODO(Nathan): blob is not ready now, it will cause panic.
-	// blobPool := blobpool.New(config.BlobPool, eth.blockchain)
+	blobPool := blobpool.New(config.BlobPool, eth.blockchain)
 
 	if config.TxPool.Journal != "" {
 		config.TxPool.Journal = stack.ResolvePath(config.TxPool.Journal)
 	}
 	legacyPool := legacypool.New(config.TxPool, eth.blockchain)
 
-	// TODO(Nathan): eth.txPool, err = txpool.New(new(big.Int).SetUint64(config.TxPool.PriceLimit), eth.blockchain, []txpool.SubPool{legacyPool, blobPool})
-	eth.txPool, err = txpool.New(new(big.Int).SetUint64(config.TxPool.PriceLimit), eth.blockchain, []txpool.SubPool{legacyPool})
+	eth.txPool, err = txpool.New(new(big.Int).SetUint64(config.TxPool.PriceLimit), eth.blockchain, []txpool.SubPool{legacyPool, blobPool})
 	if err != nil {
 		return nil, err
 	}
@@ -302,7 +301,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		Chain:                  eth.blockchain,
 		TxPool:                 eth.txPool,
 		Merger:                 eth.merger,
-		Network:                config.NetworkId,
+		Network:                networkID,
 		Sync:                   config.SyncMode,
 		BloomCache:             uint64(cacheLimit),
 		EventMux:               eth.eventMux,

eth/handler.go

@@ -990,4 +990,7 @@ func (h *handler) enableSyncedFeatures() {
 		log.Info("Snap sync complete, auto disabling")
 		h.snapSync.Store(false)
 	}
+	// if h.chain.TrieDB().Scheme() == rawdb.PathScheme {
+	// 	h.chain.TrieDB().SetBufferSize(pathdb.DefaultBufferSize)
+	// }
 }

eth/peerset.go

@@ -289,6 +289,12 @@ func (ps *peerSet) waitTrustExtension(peer *eth.Peer) (*trust.Peer, error) {
 		delete(ps.trustWait, id)
 		ps.lock.Unlock()
 		return nil, errPeerWaitTimeout
+
+	case <-ps.quitCh:
+		ps.lock.Lock()
+		delete(ps.trustWait, id)
+		ps.lock.Unlock()
+		return nil, errPeerSetClosed
 	}
 }
 
@@ -348,6 +354,12 @@ func (ps *peerSet) waitBscExtension(peer *eth.Peer) (*bsc.Peer, error) {
 				}
 			}
 		}
+
+	case <-ps.quitCh:
+		ps.lock.Lock()
+		delete(ps.bscWait, id)
+		ps.lock.Unlock()
+		return nil, errPeerSetClosed
 	}
 }
 

miner/worker.go

@@ -1127,7 +1127,7 @@ LOOP:
 		sub := w.eth.TxPool().SubscribeTransactions(txsCh, true)
 		// if TxPool has been stopped, `sub` would be nil, it could happen on shutdown.
 		if sub == nil {
-			log.Info("commitWork SubscribeNewTxsEvent return nil")
+			log.Info("commitWork SubscribeTransactions return nil")
 		} else {
 			defer sub.Unsubscribe()
 		}