Prefetch Transactions from Pool & PIP-66 Back (#2031)

* enabling pip-66 back and wait moved from seal to prepare

* bring subsecond extra time

* prefetch from pool

* more prefetch metrics

* address early announcement for rio blocks

* separate worker cache metrics

* optinal flag to disable prefetch

* prefetch gas limit flag

* tests and lint

* small fixes

* remove logs

* address duplicates

* small fix on integration test

* prefetch coverage histogram and fmt

* verify headers coverage and bor test resiliency

* address duplicates

* duplicates and lint

* addressing comments

* small fix

* address comments

* fix push tx for rpc nodes

* rename meter and intermediateroot prefetch

* address lint

* address succession number check

* fixing concurrency issues and tests for prefetch state being thrown away by gc

* make lint

* e2e worker tests

* benchmark tests

* worker tests fixed and interrupt watch while waiting

* remove parallel from tests

Author: lucca30

accounts/abi/bind/v2/dep_tree_test.go

@@ -287,7 +287,7 @@ func TestContractLinking(t *testing.T) {
 			},
 		},
 		// test two contracts can be deployed which don't share deps
-		linkTestCaseInput{
+		{
 			map[rune][]rune{
 				'a': {'b', 'c', 'd', 'e'},
 				'f': {'g', 'h', 'i', 'j'}},
@@ -297,7 +297,7 @@ func TestContractLinking(t *testing.T) {
 			},
 		},
 		// test two contracts can be deployed which share deps
-		linkTestCaseInput{
+		{
 			map[rune][]rune{
 				'a': {'b', 'c', 'd', 'e'},
 				'f': {'g', 'c', 'd', 'h'}},
@@ -307,31 +307,31 @@ func TestContractLinking(t *testing.T) {
 			},
 		},
 		// test one contract with overrides for all lib deps
-		linkTestCaseInput{
+		{
 			map[rune][]rune{
 				'a': {'b', 'c', 'd', 'e'}},
 			map[rune]struct{}{'b': {}, 'c': {}, 'd': {}, 'e': {}},
 			map[rune]struct{}{
 				'a': {}},
 		},
 		// test one contract with overrides for some lib deps
-		linkTestCaseInput{
+		{
 			map[rune][]rune{
 				'a': {'b', 'c'}},
 			map[rune]struct{}{'b': {}, 'c': {}},
 			map[rune]struct{}{
 				'a': {}},
 		},
 		// test deployment of a contract with overrides
-		linkTestCaseInput{
+		{
 			map[rune][]rune{
 				'a': {}},
 			map[rune]struct{}{'a': {}},
 			map[rune]struct{}{},
 		},
 		// two contracts ('a' and 'f') share some dependencies.  contract 'a' is marked as an override.  expect that any of
 		// its dependencies that aren't shared with 'f' are not deployed.
-		linkTestCaseInput{map[rune][]rune{
+		{map[rune][]rune{
 			'a': {'b', 'c', 'd', 'e'},
 			'f': {'g', 'c', 'd', 'h'}},
 			map[rune]struct{}{'a': {}},

cmd/utils/flags.go

@@ -598,7 +598,6 @@ var (
 		Usage:    "0x prefixed public address for the pending block producer (not used for actual block production)",
 		Category: flags.MinerCategory,
 	}
-
 	// Account settings
 	PasswordFileFlag = &cli.PathFlag{
 		Name:      "password",

consensus/beacon/consensus.go

@@ -336,14 +336,14 @@ func (beacon *Beacon) verifyHeaders(chain consensus.ChainHeaderReader, headers [
 
 // Prepare implements consensus.Engine, initializing the difficulty field of a
 // header to conform to the beacon protocol. The changes are done inline.
-func (beacon *Beacon) Prepare(chain consensus.ChainHeaderReader, header *types.Header) error {
+func (beacon *Beacon) Prepare(chain consensus.ChainHeaderReader, header *types.Header, waitOnPrepare bool) error {
 	// Transition isn't triggered yet, use the legacy rules for preparation.
 	reached, err := IsTTDReached(chain, header.ParentHash, header.Number.Uint64()-1)
 	if err != nil {
 		return err
 	}
 	if !reached {
-		return beacon.ethone.Prepare(chain, header)
+		return beacon.ethone.Prepare(chain, header, waitOnPrepare)
 	}
 	header.Difficulty = beaconDifficulty
 	return nil

consensus/bor/bor.go

@@ -402,13 +402,30 @@ func (c *Bor) verifyHeader(chain consensus.ChainHeaderReader, header *types.Head
 	number := header.Number.Uint64()
 	now := uint64(time.Now().Unix())
 
-	// Allow early blocks if Bhilai HF is enabled
-	if c.config.IsBhilai(header.Number) {
+	if c.config.IsRio(header.Number) {
+		// Rio HF introduced flexible blocktime (can be set larger than consensus without approval).
+		// Using strict CalcProducerDelay would reject valid blocks, so we just ensure announcement
+		// time comes after parent time to allow for flexible blocktime.
+		var parent *types.Header
+
+		if len(parents) > 0 {
+			parent = parents[len(parents)-1]
+		} else {
+			parent = chain.GetHeader(header.ParentHash, number-1)
+		}
+		if parent == nil || now < parent.Time {
+			log.Error("Block announced too early post rio", "number", number, "headerTime", header.Time, "now", now)
+			return consensus.ErrFutureBlock
+		}
+	} else if c.config.IsBhilai(header.Number) {
+		// Allow early blocks if Bhilai HF is enabled
 		// Don't waste time checking blocks from the future but allow a buffer of block time for
 		// early block announcements. Note that this is a loose check and would allow early blocks
 		// from non-primary producer. Such blocks will be rejected later when we know the succession
 		// number of the signer in the current sprint.
-		if header.Time-c.config.CalculatePeriod(number) > now {
+		// Uses CalcProducerDelay instead of block period to account for producer delay on sprint start blocks.
+		// We assume succession 0 (primary producer) to not be much restrictive for early block announcements.
+		if header.Time-CalcProducerDelay(number, 0, c.config) > now {
 			log.Error("Block announced too early post bhilai", "number", number, "headerTime", header.Time, "now", now)
 			return consensus.ErrFutureBlock
 		}
@@ -476,7 +493,18 @@ func (c *Bor) verifyHeader(chain consensus.ChainHeaderReader, header *types.Head
 		return err
 	}
 
-	c.recentVerifiedHeaders.Set(header.Hash(), header, ttlcache.DefaultTTL)
+	// Calculate TTL for the header cache entry
+	// If the header time is in the future (early announced block), add extra time to TTL
+	cacheTTL := veblopBlockTimeout
+	nowTime := time.Now()
+	headerTime := time.Unix(int64(header.Time), 0)
+	if headerTime.After(nowTime) {
+		// Add the time from now until header time as extra to the base timeout
+		extraTime := headerTime.Sub(nowTime)
+		cacheTTL = veblopBlockTimeout + extraTime
+	}
+
+	c.recentVerifiedHeaders.Set(header.Hash(), header, cacheTTL)
 	return nil
 }
 
@@ -928,7 +956,7 @@ func IsBlockEarly(parent *types.Header, header *types.Header, number uint64, suc
 
 // Prepare implements consensus.Engine, preparing all the consensus fields of the
 // header for running the transactions on top.
-func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header) error {
+func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header, waitOnPrepare bool) error {
 	// If the block isn't a checkpoint, cast a random vote (good enough for now)
 	header.Coinbase = common.Address{}
 	header.Nonce = types.BlockNonce{}
@@ -1026,6 +1054,8 @@ func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header) e
 		return fmt.Errorf("the floor of custom mining block time (%v) is less than the consensus block time: %v < %v", c.blockTime, c.blockTime.Seconds(), c.config.CalculatePeriod(number))
 	}
 
+	var delay time.Duration
+
 	if c.blockTime > 0 && c.config.IsRio(header.Number) {
 		// Only enable custom block time for Rio and later
 
@@ -1043,14 +1073,16 @@ func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header) e
 		actualNewBlockTime := parentActualBlockTime.Add(c.blockTime)
 		header.Time = uint64(actualNewBlockTime.Unix())
 		header.ActualTime = actualNewBlockTime
+		delay = time.Until(parentActualBlockTime)
 	} else {
 		header.Time = parent.Time + CalcProducerDelay(number, succession, c.config)
+		delay = time.Until(time.Unix(int64(parent.Time), 0))
 	}
 
 	now := time.Now()
 	if header.Time < uint64(now.Unix()) {
 		additionalBlockTime := time.Duration(c.config.CalculatePeriod(number)) * time.Second
-		if c.blockTime > 0 {
+		if c.blockTime > 0 && c.config.IsRio(header.Number) {
 			additionalBlockTime = c.blockTime
 		}
 		header.Time = uint64(now.Add(additionalBlockTime).Unix())
@@ -1059,6 +1091,22 @@ func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header) e
 		}
 	}
 
+	// Wait before start the block production if needed (previsously this wait was on Seal)
+	if c.config.IsBhilai(header.Number) && waitOnPrepare {
+		var successionNumber int
+		// if signer is not empty (RPC nodes have empty signer)
+		if currentSigner.signer != (common.Address{}) {
+			var err error
+			successionNumber, err = snap.GetSignerSuccessionNumber(currentSigner.signer)
+			if err != nil {
+				return err
+			}
+			if successionNumber == 0 {
+				<-time.After(delay)
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -1323,14 +1371,8 @@ func (c *Bor) Seal(chain consensus.ChainHeaderReader, block *types.Block, witnes
 	var delay time.Duration
 
 	// Sweet, the protocol permits us to sign the block, wait for our time
-	if c.config.IsBhilai(header.Number) {
-		delay = time.Until(header.GetActualTime()) // Wait until we reach header time for non-primary validators
-		// Disable early block announcement
-		// if successionNumber == 0 {
-		// 	// For primary producers, set the delay to `header.Time - block time` instead of `header.Time`
-		// 	// for early block announcement instead of waiting for full block time.
-		// 	delay = time.Until(time.Unix(int64(header.Time-c.config.CalculatePeriod(number)), 0))
-		// }
+	if c.config.IsBhilai(header.Number) && successionNumber == 0 {
+		delay = 0 // delay was moved to Prepare for bhilai and later
 	} else {
 		delay = time.Until(header.GetActualTime()) // Wait until we reach header time
 	}