Merge pull request #302 from bane-labs/fix-persist

dbft: untie BlockQueue from worker

Author: AnnaShaleva

consensus/dbft/blockqueue.go

@@ -1,47 +1,26 @@
 package dbft
 
 import (
-	"errors"
 	"fmt"
-	"github.com/ethereum/go-ethereum/core/state"
-	"sync"
 
-	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 )
 
-const (
-	// blockQueueCap is the number of tasks blockQueue can fit at once. It's OK for
-	// the blockQueue not to have a proper task for the newly-created block, and
-	// normally a single task is expected to be present in blockQueue. But we still
-	// need blockQueueCap restriction for the case of endless change views.
-	blockQueueCap = 100
-
-	clearAllMatchingTasks = -1
-)
-
-// blockQueue is an entity that collects sealed blocks from dBFT and routs these
-// blocks to a proper place (either to miner or directly to chain).
+// blockQueue is an entity that collects sealed blocks from dBFT and stores these
+// blocks in the chain.
 type blockQueue struct {
-	chain     ChainHeaderWriter
-	tasksLock sync.RWMutex
-	tasks     map[common.Hash]task
-}
-
-// task holds information about miner sealing task.
-type task struct {
-	height   uint64
-	resCh    chan<- *types.Block
-	cancelCh <-chan struct{}
+	chain ChainHeaderWriter
+	mux   *event.TypeMux
 }
 
 // newBlockQueue creates an instance of blockQueue. It's not ready for usage until
 // an instance of ChainHeaderWriter is properly set.
 func newBlockQueue() *blockQueue {
-	return &blockQueue{
-		tasks: make(map[common.Hash]task),
-	}
+	return &blockQueue{}
 }
 
 // SetChain initializes ChainHeaderWriter instanse needed for proper blockQueue
@@ -50,54 +29,15 @@ func (bq *blockQueue) SetChain(chain ChainHeaderWriter) {
 	bq.chain = chain
 }
 
+// SetMux initializes mux instanse needed for proper blockQueue functioning.
+func (bq *blockQueue) SetMux(mux *event.TypeMux) {
+	bq.mux = mux
+}
+
 // PutBlock routs block either to miner or (if there's no suitable sealing task)
 // directly to blockchain. No block verification is performed, it is assumed that
 // provided block is sealed and valid.
 func (bq *blockQueue) PutBlock(b *types.Block, state *state.StateDB, receipts []*types.Receipt) error {
-	h := WorkerSealHash(b.Header())
-
-	bq.tasksLock.Lock()
-	task, ok := bq.tasks[h]
-
-	bq.clearStaleTasks(b.NumberU64(), clearAllMatchingTasks)
-
-	if ok {
-		var (
-			err         error
-			readByMiner bool
-		)
-		select {
-		case <-task.cancelCh:
-		case task.resCh <- b:
-			readByMiner = true
-		default:
-			err = errors.New("sealing result is not read by miner, trying to insert block in chain manually")
-		}
-		delete(bq.tasks, h)
-
-		if readByMiner {
-			bq.tasksLock.Unlock()
-			return nil
-		}
-
-		if err != nil {
-			log.Warn(err.Error(),
-				"number", b.Number(),
-				"seal hash", h.String(),
-				"hash", b.Hash().String(),
-			)
-		}
-	}
-	bq.tasksLock.Unlock()
-
-	// If we're here then we're OK with that, it just means that:
-	//  1) either dBFT received some extra commits and trying to
-	//     send already constructed block one more time
-	//  2) or worker has received block with the same index via network. Then
-	//     we still need to save the block in case it has different hash.
-	//  3) or we're not a primary node in this consensus round and thus,
-	//     worker's task differs from the dBFT's proposal. In this case we
-	//     need to try to insert block right into chain.
 	hash := b.Hash()
 	if bq.chain.HasBlock(hash, b.NumberU64()) {
 		return nil
@@ -110,9 +50,14 @@ func (bq *blockQueue) PutBlock(b *types.Block, state *state.StateDB, receipts []
 			return fmt.Errorf("failed to insert block into chain: %w", err)
 		}
 		log.Info("Successfully inserted new block", "number", b.Number(), "hash", hash)
+
+		// Broadcast the block and announce chain insertion event
+		bq.mux.Post(core.NewMinedBlockEvent{Block: b})
+		
 		return nil
 	}
 
+	currH := bq.chain.CurrentBlock().Number
 	// Insert state directly if we have one.
 	var logs []*types.Log
 	for i, receipt := range receipts {
@@ -129,7 +74,7 @@ func (bq *blockQueue) PutBlock(b *types.Block, state *state.StateDB, receipts []
 		logs = append(logs, receipt.Logs...)
 	}
 	// Commit block and state to database.
-	_, err := bq.chain.WriteBlockAndSetHead(b, receipts, logs, state, true)
+	_, err := bq.chain.WriteBlockAndSetHead(b, receipts, logs, state, b.Number().Cmp(currH) > 0)
 	if err != nil {
 		log.Error("Failed to write block to chain and set head",
 			"number", b.NumberU64(),
@@ -138,51 +83,8 @@ func (bq *blockQueue) PutBlock(b *types.Block, state *state.StateDB, receipts []
 	}
 	log.Info("Successfully wrote new block with state", "number", b.Number(), "hash", hash)
 
-	return nil
-}
-
-// ClearStaleTasks removes all stale tasks up to the specified height (including
-// the height itself).
-func (bq *blockQueue) ClearStaleTasks(till uint64) {
-	bq.tasksLock.Lock()
-	defer bq.tasksLock.Unlock()
-
-	bq.clearStaleTasks(till, clearAllMatchingTasks)
-}
-
-// clearStaleTasks removes all stale tasks up to the specified height (including
-// the height itself). It doesn't hold tasksLock, so it's the caller's responsibility.
-func (bq *blockQueue) clearStaleTasks(till uint64, count int) {
-	for h, task := range bq.tasks {
-		if task.height <= till {
-			delete(bq.tasks, h)
-			if count != clearAllMatchingTasks {
-				count--
-				if count <= 0 {
-					break
-				}
-			}
-		}
-	}
-}
-
-// SubmitTask adds subsequent miner task to the blockqueue instance.
-func (bq *blockQueue) SubmitTask(sealHash common.Hash, number uint64, resCh chan<- *types.Block, cancelCh <-chan struct{}) {
-	bq.tasksLock.Lock()
-	defer bq.tasksLock.Unlock()
+	// Broadcast the block and announce chain insertion event
+	bq.mux.Post(core.NewMinedBlockEvent{Block: b})
 
-	// We're OK with the fact that capacity is reached, remove random outdated seal
-	// task (it's likely won't be completed, and if it will, then the block will be
-	// inserted to the chain directly).
-	if len(bq.tasks) == blockQueueCap {
-		bq.clearStaleTasks(number, 1)
-	}
-
-	// Do not check the existing task with the same hash. It could happen that new
-	// sealing task has the same hash after ChangeView sealing proposal initialisation.
-	bq.tasks[sealHash] = task{
-		height:   number,
-		resCh:    resCh,
-		cancelCh: cancelCh,
-	}
+	return nil
 }

consensus/dbft/dbft.go

@@ -356,6 +356,8 @@ func New(config *params.DBFTConfig, _ ethdb.Database) (*DBFT, error) {
 					log.Warn("error on enqueue block", "error", err.Error())
 				}
 			}
+
+			c.postBlock(res.Header())
 		}),
 		dbft.WithNewBlockFromContext[common.Hash](func(ctx *dbft.Context[common.Hash]) dbft.Block[common.Hash] {
 			prepareReq := ctx.PreparationPayloads[ctx.PrimaryIndex]
@@ -714,6 +716,7 @@ func (c *DBFT) WithRequestTxs(f func(hashed []common.Hash)) {
 // the ongoing node sync process.
 func (c *DBFT) WithMux(mux *event.TypeMux) {
 	c.mux = mux
+	c.blockQueue.SetMux(mux)
 
 	go c.syncWatcher()
 }
@@ -765,9 +768,7 @@ func (c *DBFT) WithTxPool(pool txPool) {
 
 // postBlock is a callback that updates latest accepted block data and resets
 // last proposal data. It must be called every time new block arrives from chain
-// or from consensus. It must be called strictly after new block persist since
-// NextConsensus calculation depends on the storage state. It also clears all
-// BlockQueue tasks up to the accepted block height.
+// or from consensus.
 func (c *DBFT) postBlock(h *types.Header) {
 	num := h.Number.Uint64()
 	if c.lastIndex < num {
@@ -776,8 +777,6 @@ func (c *DBFT) postBlock(h *types.Header) {
 		c.lastBlockHash = h.Hash()
 		c.lastBlockSealHash = HonestSealHash(h)
 		c.lastBlockExtra = h.Extra
-
-		c.blockQueue.ClearStaleTasks(num)
 	}
 }
 
@@ -1194,9 +1193,6 @@ func (c *DBFT) Seal(chain consensus.ChainHeaderReader, b *types.Block, results c
 	c.lastProposal = b
 	c.lastProposalLock.Unlock()
 
-	sealHash := c.SealHash(b.Header())
-	c.blockQueue.SubmitTask(sealHash, b.NumberU64(), results, stop)
-
 	return nil
 }
 

eth/protocols/dbft/handler_test.go

@@ -29,9 +29,9 @@ func TestHandling(t *testing.T) {
 		key, _         = crypto.GenerateKey()
 		addr           = crypto.PubkeyToAddress(key.PublicKey)
 		bc             = &testBC{height: 10}
-		s1             = New(bc, nil, func(u uint64, address common.Address) bool { return true })
-		s2             = New(bc, nil, func(u uint64, address common.Address) bool { return true })
-		s3             = New(bc, nil, func(u uint64, address common.Address) bool { return true })
+		s1             = New(bc, nil, func(u uint64, address common.Address) error { return nil })
+		s2             = New(bc, nil, func(u uint64, address common.Address) error { return nil })
+		s3             = New(bc, nil, func(u uint64, address common.Address) error { return nil })
 		p1             = p2p.NewPeer(enode.ID{1}, "peer1", nil)
 		p2             = p2p.NewPeer(enode.ID{2}, "peer2", nil)
 		p3             = p2p.NewPeer(enode.ID{3}, "peer3", nil)

eth/protocols/dbft/pool.go

@@ -6,6 +6,7 @@ package dbft
 import (
 	"container/list"
 	"errors"
+	"fmt"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -96,7 +97,7 @@ func (p *Pool) verify(m *Message) (bool, error) {
 		if errors.Is(err, ErrSyncing) {
 			return false, nil
 		}
-		return false, err
+		return false, fmt.Errorf("%w: %w", errDisallowedSender, err)
 	}
 	return true, nil
 }

eth/protocols/dbft/pool_test.go

@@ -22,6 +22,12 @@ func TestAddGet(t *testing.T) {
 		m := someMessage(t, 100, bc.goodAddrs[0], bc.badKey)
 		p.testAdd(t, false, invalidSig, m)
 	})
+	t.Run("syncing", func(t *testing.T) {
+		bc.syncing = true
+		m := bc.badMessage(t, 100)
+		p.testAdd(t, false, nil, m)
+		bc.syncing = false
+	})
 	t.Run("disallowed sender", func(t *testing.T) {
 		m := bc.badMessage(t, 100)
 		p.testAdd(t, false, errDisallowedSender, m)
@@ -106,6 +112,7 @@ type testChain struct {
 	badKey    *ecdsa.PrivateKey
 	goodAddrs []common.Address
 	badAddr   common.Address
+	syncing   bool
 }
 
 var errVerification = errors.New("verification failed")
@@ -120,13 +127,16 @@ func newTestChain() *testChain {
 		badAddr:   crypto.PubkeyToAddress(bk.PublicKey),
 	}
 }
-func (c *testChain) IsAddressAllowed(u common.Address) bool {
+func (c *testChain) IsAddressAllowed(u common.Address) error {
+	if c.syncing {
+		return ErrSyncing
+	}
 	for i := range c.goodAddrs {
 		if u == c.goodAddrs[i] {
-			return true
+			return nil
 		}
 	}
-	return false
+	return errors.New("address not allowed")
 }
 func (c *testChain) BlockHeight() uint64 { return c.height }