consensus, core, eth, miner: use the same insert behavior among validators

Author: markya0616

consensus/consensus.go

@@ -113,8 +113,11 @@ type Istanbul interface {
 	NewChainHead(block *types.Block) error
 
 	// Start starts the engine
-	Start(chain ChainReader, inserter func(types.Blocks) (int, error)) error
+	Start(chain ChainReader) error
 
 	// Stop stops the engine
 	Stop() error
+
+	// Set commit and write proposed work functions
+	SetWorkerFns(func(*types.Block) error, func(*types.Block) bool)
 }

consensus/istanbul/backend/backend.go

@@ -27,6 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
 	istanbulCore "github.com/ethereum/go-ethereum/consensus/istanbul/core"
 	"github.com/ethereum/go-ethereum/consensus/istanbul/validator"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
@@ -69,7 +70,10 @@ type backend struct {
 	logger           log.Logger
 	db               ethdb.Database
 	chain            consensus.ChainReader
-	inserter         func(types.Blocks) (int, error)
+	// commitProposedWork tries to commit the transactions in proposed block into chain DB
+	commitProposedWork func(*types.Block) error
+	// writeProposedWork tries to write the proposed block into chain DB
+	writeProposedWork func(*types.Block) bool
 
 	// the channels for istanbul engine notifications
 	commitCh          chan *types.Block
@@ -151,12 +155,13 @@ func (sb *backend) Commit(proposal istanbul.Proposal, seals [][]byte) error {
 	if sb.proposedBlockHash == block.Hash() {
 		// feed block hash to Seal() and wait the Seal() result
 		sb.commitCh <- block
-		// TODO: how do we check the block is inserted correctly?
+		go sb.eventMux.Post(core.NewMinedBlockEvent{Block: block})
 		return nil
 	}
-	// if I'm not a proposer, insert the block directly and broadcast NewCommittedEvent
-	if _, err := sb.inserter(types.Blocks{block}); err != nil {
-		return err
+	if result := sb.writeProposedWork(block); !result {
+		// if I'm not a proposer, insert the block directly and broadcast NewCommittedEvent
+		sb.logger.Warn("Write work failed")
+		return nil
 	}
 	msg := istanbul.NewCommittedEvent{
 		Block: block,
@@ -190,12 +195,12 @@ func (sb *backend) Verify(proposal istanbul.Proposal) (time.Duration, error) {
 	// verify the header of proposed block
 	err := sb.VerifyHeader(sb.chain, block.Header(), false)
 	// ignore errEmptyCommittedSeals error because we don't have the committed seals yet
-	if err == nil || err == errEmptyCommittedSeals {
-		return 0, nil
-	} else if err == consensus.ErrFutureBlock {
+	if err == consensus.ErrFutureBlock {
 		return time.Unix(block.Header().Time.Int64(), 0).Sub(now()), consensus.ErrFutureBlock
+	} else if err != nil && err != errEmptyCommittedSeals {
+		return 0, err
 	}
-	return 0, err
+	return 0, sb.commitProposedWork(block)
 }
 
 // Sign implements istanbul.Backend.Sign

consensus/istanbul/backend/backend_test.go

@@ -29,7 +29,8 @@ import (
 	"github.com/ethereum/go-ethereum/consensus/istanbul/validator"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/event"
 )
 
 func TestSign(t *testing.T) {
@@ -265,10 +266,10 @@ func (slice Keys) Swap(i, j int) {
 func newBackend() (b *backend, validatorKeys Keys, validatorSet istanbul.ValidatorSet) {
 	key, _ := generatePrivateKey()
 	validatorSet, validatorKeys = newTestValidatorSet(5)
-	b = &backend{
-		privateKey: key,
-		logger:     log.New("backend", "simple"),
-		commitCh:   make(chan *types.Block, 1),
-	}
+	eventMux := new(event.TypeMux)
+	memDB, _ := ethdb.NewMemDatabase()
+	config := istanbul.DefaultConfig
+	// Use the first key as private key
+	b, _ = New(config, eventMux, key, memDB).(*backend)
 	return
 }

consensus/istanbul/backend/engine.go

@@ -531,7 +531,7 @@ func (sb *backend) NewChainHead(block *types.Block) error {
 }
 
 // Start implements consensus.Istanbul.Start
-func (sb *backend) Start(chain consensus.ChainReader, inserter func(types.Blocks) (int, error)) error {
+func (sb *backend) Start(chain consensus.ChainReader) error {
 	sb.coreMu.Lock()
 	defer sb.coreMu.Unlock()
 	if sb.coreStarted {
@@ -546,7 +546,6 @@ func (sb *backend) Start(chain consensus.ChainReader, inserter func(types.Blocks
 	sb.commitCh = make(chan *types.Block, 1)
 
 	sb.chain = chain
-	sb.inserter = inserter
 
 	curHeader := chain.CurrentHeader()
 	lastSequence := new(big.Int).Set(curHeader.Number)
@@ -583,6 +582,11 @@ func (sb *backend) Stop() error {
 	return nil
 }
 
+func (sb *backend) SetWorkerFns(commitFn func(*types.Block) error, writeFn func(*types.Block) bool) {
+	sb.commitProposedWork = commitFn
+	sb.writeProposedWork = writeFn
+}
+
 // snapshot retrieves the authorization snapshot at a given point in time.
 func (sb *backend) snapshot(chain consensus.ChainReader, number uint64, hash common.Hash, parents []*types.Header) (*Snapshot, error) {
 	// Search for a snapshot in memory or on disk for checkpoints

consensus/istanbul/backend/engine_test.go

@@ -53,7 +53,7 @@ func newBlockChain(n int) (*core.BlockChain, *backend) {
 	if err != nil {
 		panic(err)
 	}
-	b.Start(blockchain, blockchain.InsertChain)
+	b.Start(blockchain)
 	snap, err := b.snapshot(blockchain, 0, common.Hash{}, nil)
 	if err != nil {
 		panic(err)

consensus/istanbul/core/preprepare.go

@@ -68,21 +68,23 @@ func (c *core) handlePreprepare(msg *message, src istanbul.Validator) error {
 	}
 
 	// Verify the proposal we received
-	if duration, err := c.backend.Verify(preprepare.Proposal); err != nil {
-		logger.Warn("Failed to verify proposal", "err", err, "duration", duration)
-		// if it's a future block, we will handle it again after the duration
-		if err == consensus.ErrFutureBlock {
-			c.stopFuturePreprepareTimer()
-			c.futurePreprepareTimer = time.AfterFunc(duration, func() {
-				c.sendEvent(backlogEvent{
-					src: src,
-					msg: msg,
+	if !c.isProposer() {
+		if duration, err := c.backend.Verify(preprepare.Proposal); err != nil {
+			logger.Warn("Failed to verify proposal", "err", err, "duration", duration)
+			// if it's a future block, we will handle it again after the duration
+			if err == consensus.ErrFutureBlock {
+				c.stopFuturePreprepareTimer()
+				c.futurePreprepareTimer = time.AfterFunc(duration, func() {
+					c.sendEvent(backlogEvent{
+						src: src,
+						msg: msg,
+					})
 				})
-			})
-		} else {
-			c.sendNextRoundChange()
+			} else {
+				c.sendNextRoundChange()
+			}
+			return err
 		}
-		return err
 	}
 
 	// Here is about to accept the preprepare

consensus/istanbul/events.go

@@ -31,7 +31,9 @@ type ConsensusDataEvent struct {
 
 // NewCommittedEvent is posted when I'm not a proposer but
 // a block has been committed from Istanbul consensus.
-type NewCommittedEvent struct{ Block *types.Block }
+type NewCommittedEvent struct {
+	Block *types.Block
+}
 
 // RequestEvent is posted to propose a proposal
 type RequestEvent struct {

core/blockchain.go

@@ -827,6 +827,11 @@ func (bc *BlockChain) WriteBlock(block *types.Block) (status WriteStatus, err er
 	bc.mu.Lock()
 	defer bc.mu.Unlock()
 
+	if bc.HasBlock(block.Hash()) {
+		log.Debug("Block exists", "hash", block.Hash())
+		return
+	}
+
 	localTd := bc.GetTd(bc.currentBlock.Hash(), bc.currentBlock.NumberU64())
 	externTd := new(big.Int).Add(block.Difficulty(), ptd)
 

eth/backend.go

@@ -355,7 +355,7 @@ func (s *Ethereum) StartMining(local bool) error {
 		}
 		clique.Authorize(eb, wallet.SignHash)
 	} else if istanbul, ok := s.engine.(consensus.Istanbul); ok {
-		istanbul.Start(s.blockchain, s.blockchain.InsertChain)
+		istanbul.Start(s.blockchain)
 	}
 	if local {
 		// If local (CPU) mining is started, we can disable the transaction rejection