core: fix empty parent not found (#470)

* Do not panic when empty block is unable to propose
* Make blockChain able to handle pulled empty block

3 files changed, 61 insertions, 22 deletions

diff --git a/core/blockchain.go b/core/blockchain.go
index 30895f8..aacb65d 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -63,6 +63,13 @@ func (pb *pendingBlockRecords) insert(p pendingBlockRecord) error {
 		*pb = append(*pb, p)
 	default:
 		if (*pb)[idx].position.Equal(p.position) {
+			// Allow to overwrite pending block record for empty blocks, we may
+			// need to pull that block from others when its parent is not found
+			// locally.
+			if (*pb)[idx].block == nil && p.block != nil {
+				(*pb)[idx].block = p.block
+				return nil
+			}
 			return ErrDuplicatedPendingBlock
 		}
 		// Insert the value to that index.
@@ -275,8 +282,7 @@ func (bc *blockChain) addEmptyBlock(position types.Position) (
 	} else if position.Height == 0 && position.Round == 0 {
 		return add(), nil
 	}
-	bc.addPendingBlockRecord(pendingBlockRecord{position, nil})
-	return nil, nil
+	return nil, bc.addPendingBlockRecord(pendingBlockRecord{position, nil})
 }
 
 // addBlock should be called when the block is confirmed by BA, we won't perform
@@ -298,11 +304,10 @@ func (bc *blockChain) addBlock(b *types.Block) error {
 		confirmed = true
 	}
 	if !confirmed {
-		bc.addPendingBlockRecord(pendingBlockRecord{b.Position, b})
-	} else {
-		bc.confirmBlock(b)
-		bc.checkIfBlocksConfirmed()
+		return bc.addPendingBlockRecord(pendingBlockRecord{b.Position, b})
 	}
+	bc.confirmBlock(b)
+	bc.checkIfBlocksConfirmed()
 	return nil
 }
 
@@ -449,18 +454,19 @@ func (bc *blockChain) findPendingBlock(p types.Position) *types.Block {
 	return pendingRec.block
 }
 
-func (bc *blockChain) addPendingBlockRecord(p pendingBlockRecord) {
+func (bc *blockChain) addPendingBlockRecord(p pendingBlockRecord) error {
 	if err := bc.pendingBlocks.insert(p); err != nil {
 		if err == ErrDuplicatedPendingBlock {
-			// TODO(mission): panic directly once our BA can confirm blocks
-			//                uniquely and in sequence.
-			return
+			// TODO(mission): stop ignoreing this error once our BA can confirm
+			//                blocks uniquely and sequentially.
+			err = nil
 		}
-		panic(err)
+		return err
 	}
 	if p.block != nil {
 		bc.setRandomnessFromPending(p.block)
 	}
+	return nil
 }
 
 func (bc *blockChain) checkIfBlocksConfirmed() {
diff --git a/core/blockchain_test.go b/core/blockchain_test.go
index 4c8fe23..991dc94 100644
--- a/core/blockchain_test.go
+++ b/core/blockchain_test.go
@@ -419,10 +419,8 @@ func (s *BlockChainTestSuite) TestPendingBlockRecords() {
 	s.Require().NoError(ps.insert(pendingBlockRecord{bs[2].Position, bs[2]}))
 	s.Require().NoError(ps.insert(pendingBlockRecord{bs[1].Position, bs[1]}))
 	s.Require().NoError(ps.insert(pendingBlockRecord{bs[0].Position, bs[0]}))
-	// TODO(mission): unlock this checking once our BA can confirm blocks
-	//                uniquely in sequence.
-	// s.Require().Equal(ErrDuplicatedPendingBlock.Error(),
-	//    ps.insert(pendingBlockRecord{bs[0].Position, nil}).Error())
+	s.Require().Equal(ErrDuplicatedPendingBlock.Error(),
+		ps.insert(pendingBlockRecord{bs[0].Position, nil}).Error())
 	s.Require().True(ps[0].position.Equal(bs[0].Position))
 	s.Require().True(ps[1].position.Equal(bs[1].Position))
 	s.Require().True(ps[2].position.Equal(bs[2].Position))
@@ -458,16 +456,43 @@ func (s *BlockChainTestSuite) TestAddEmptyBlockDirectly() {
 	s.Require().NoError(bc.addBlock(blocks[0]))
 	// Add an empty block after a normal block.
 	pos := types.Position{Height: 1}
-	emptyB, err := bc.addEmptyBlock(pos)
-	s.Require().NotNil(emptyB)
-	s.Require().True(emptyB.Position.Equal(pos))
+	emptyB1, err := bc.addEmptyBlock(pos)
+	s.Require().NotNil(emptyB1)
+	s.Require().True(emptyB1.Position.Equal(pos))
 	s.Require().NoError(err)
 	// Add an empty block after an empty block.
 	pos = types.Position{Height: 2}
-	emptyB, err = bc.addEmptyBlock(pos)
-	s.Require().NotNil(emptyB)
-	s.Require().True(emptyB.Position.Equal(pos))
+	emptyB2, err := bc.addEmptyBlock(pos)
+	s.Require().NotNil(emptyB2)
+	s.Require().True(emptyB2.Position.Equal(pos))
 	s.Require().NoError(err)
+	// prepare a normal block.
+	pos = types.Position{Height: 3}
+	b3, err := bc.proposeBlock(pos, emptyB2.Timestamp.Add(s.blockInterval))
+	s.Require().NotNil(b3)
+	s.Require().NoError(err)
+	// Add an empty block far away from current tip.
+	pos = types.Position{Height: 4}
+	emptyB4, err := bc.addEmptyBlock(pos)
+	s.Require().Nil(emptyB4)
+	s.Require().NoError(err)
+	// propose an empty block based on the block at height=3, which mimics the
+	// scenario that the empty block is pulled from others.
+	emptyB4 = &types.Block{
+		ParentHash: b3.Hash,
+		Position:   pos,
+		Timestamp:  b3.Timestamp.Add(s.blockInterval),
+		Witness: types.Witness{
+			Height: b3.Witness.Height,
+			Data:   b3.Witness.Data, // Hacky, don't worry.
+		},
+	}
+	emptyB4.Hash, err = utils.HashBlock(emptyB4)
+	s.Require().NoError(err)
+	s.Require().NoError(bc.addBlock(emptyB4))
+	rec, found := bc.pendingBlocks.searchByHeight(4)
+	s.Require().True(found)
+	s.Require().NotNil(rec.block)
 }
 
 func TestBlockChain(t *testing.T) {
diff --git a/core/consensus.go b/core/consensus.go
index 871e855..050bfe7 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -127,7 +127,15 @@ func (recv *consensusBAReceiver) ConfirmBlock(
 			return
 		}
 		if block == nil {
-			panic(fmt.Errorf("empty block should be proposed directly: %s", aID))
+			// The empty block's parent is not found locally, thus we can't
+			// propose it at this moment.
+			//
+			// We can only rely on block pulling upon receiving
+			// types.AgreementResult from the next position.
+			recv.consensus.logger.Warn(
+				"An empty block is confirmed without its parent",
+				"position", aID)
+			return
 		}
 	} else {
 		var exist bool