sync: coreth PR #1323: wait for tx pool event loop (#1871)

Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

plugin/evm/block_builder.go

@@ -47,6 +47,9 @@ type blockBuilder struct {
 	// but at least after a minimum delay of minBlockBuildingRetryDelay.
 	lastBuildParentHash common.Hash
 	lastBuildTime       time.Time
+
+	chainHeadHash   common.Hash
+	mempoolHeadHash common.Hash
 }
 
 func (vm *VM) NewBlockBuilder() *blockBuilder {
@@ -94,9 +97,13 @@ func (b *blockBuilder) awaitSubmittedTxs() {
 	txSubmitChan := make(chan core.NewTxsEvent)
 	b.txPool.SubscribeTransactions(txSubmitChan, true)
 
+	events := make(chan core.NewTxPoolReorgEvent)
+	sub := b.txPool.SubscribeNewReorgEvent(events)
+
 	b.shutdownWg.Add(1)
 	go b.ctx.Log.RecoverAndPanic(func() {
 		defer b.shutdownWg.Done()
+		defer sub.Unsubscribe()
 
 		for {
 			select {
@@ -105,6 +112,12 @@ func (b *blockBuilder) awaitSubmittedTxs() {
 				b.signalCanBuild()
 			case <-b.shutdownChan:
 				return
+			case event := <-events:
+				if event.Head == nil || event.Head.Number == nil {
+					log.Warn("nil head or block number in tx pool reorg event")
+					continue
+				}
+				b.setMempoolHeadHash(event.Head.Hash())
 			}
 		}
 	})
@@ -143,7 +156,7 @@ func (b *blockBuilder) waitForEvent(ctx context.Context, currentHeader *types.He
 func (b *blockBuilder) waitForNeedToBuild(ctx context.Context) (time.Time, common.Hash, error) {
 	b.buildBlockLock.Lock()
 	defer b.buildBlockLock.Unlock()
-	for !b.needToBuild() {
+	for !b.needToBuild() || b.pendingPoolUpdate() {
 		if err := b.pendingSignal.Wait(ctx); err != nil {
 			return time.Time{}, common.Hash{}, err
 		}
@@ -186,3 +199,33 @@ func minNextBlockTime(parent *types.Header) time.Time {
 	requiredDelay := time.Duration(acp226DelayExcess.Delay()) * time.Millisecond
 	return parentTime.Add(requiredDelay)
 }
+
+func (b *blockBuilder) setChainHeadHash(hash common.Hash) {
+	b.buildBlockLock.Lock()
+	defer b.buildBlockLock.Unlock()
+
+	b.chainHeadHash = hash
+
+	if b.pendingPoolUpdate() {
+		return
+	}
+
+	b.pendingSignal.Broadcast()
+}
+
+func (b *blockBuilder) setMempoolHeadHash(hash common.Hash) {
+	b.buildBlockLock.Lock()
+	defer b.buildBlockLock.Unlock()
+
+	b.mempoolHeadHash = hash
+
+	if b.pendingPoolUpdate() {
+		return
+	}
+
+	b.pendingSignal.Broadcast()
+}
+
+func (b *blockBuilder) pendingPoolUpdate() bool {
+	return b.chainHeadHash != b.mempoolHeadHash
+}

plugin/evm/vm.go

@@ -891,6 +891,8 @@ func (vm *VM) onNormalOperationsStarted() error {
 	// NOTE: gossip network must be initialized first otherwise ETH tx gossip will not work.
 	vm.builderLock.Lock()
 	vm.builder = vm.NewBlockBuilder()
+	vm.builder.setChainHeadHash(vm.blockChain.CurrentBlock().Hash())
+	vm.builder.setMempoolHeadHash(vm.blockChain.CurrentBlock().Hash())
 	vm.builder.awaitSubmittedTxs()
 	vm.builderLock.Unlock()
 
@@ -1130,7 +1132,14 @@ func (vm *VM) SetPreference(ctx context.Context, blkID ids.ID) error {
 		return fmt.Errorf("failed to set preference to %s: %w", blkID, err)
 	}
 
-	return vm.blockChain.SetPreference(block.(*wrappedBlock).ethBlock)
+	wb, isWrappedBlock := block.(*wrappedBlock)
+	if !isWrappedBlock {
+		return fmt.Errorf("expected block %s to be of type *wrappedBlock but got %T", blkID, block)
+	}
+
+	vm.setPendingBlock(wb.GetEthBlock().Hash())
+
+	return vm.blockChain.SetPreference(wb.ethBlock)
 }
 
 // GetBlockIDAtHeight returns the canonical block at [height].
@@ -1370,6 +1379,17 @@ func attachEthService(handler *rpc.Server, apis []rpc.API, names []string) error
 	return nil
 }
 
+func (vm *VM) setPendingBlock(hash common.Hash) {
+	vm.builderLock.Lock()
+	defer vm.builderLock.Unlock()
+
+	if vm.builder == nil {
+		return
+	}
+
+	vm.builder.setChainHeadHash(hash)
+}
+
 func (vm *VM) Connected(ctx context.Context, nodeID ids.NodeID, version *version.Application) error {
 	vm.vmLock.Lock()
 	defer vm.vmLock.Unlock()