core: deliver finalized blocks upon receiving randomness results. (#376)

6 files changed, 47 insertions, 21 deletions

diff --git a/core/compaction-chain.go b/core/compaction-chain.go
index 6192abd..14e3b26 100644
--- a/core/compaction-chain.go
+++ b/core/compaction-chain.go
@@ -125,8 +125,6 @@ func (cc *compactionChain) processBlock(block *types.Block) error {
 }
 
 func (cc *compactionChain) extractBlocks() []*types.Block {
-	prevBlock := cc.lastDeliveredBlock()
-
 	// Check if we're synced.
 	if !func() bool {
 		cc.lock.RLock()
@@ -135,7 +133,7 @@ func (cc *compactionChain) extractBlocks() []*types.Block {
 			return false
 		}
 		// Finalization.Height == 0 is syncing from bootstrap.
-		if prevBlock.Finalization.Height == 0 {
+		if cc.prevBlock.Finalization.Height == 0 {
 			return cc.chainUnsynced == 0
 		}
 		return true
@@ -145,7 +143,10 @@ func (cc *compactionChain) extractBlocks() []*types.Block {
 	deliveringBlocks := make([]*types.Block, 0)
 	cc.lock.Lock()
 	defer cc.lock.Unlock()
-	var block *types.Block
+	var (
+		block     *types.Block
+		prevBlock = cc.prevBlock
+	)
 	for len(cc.pendingBlocks) > 0 &&
 		(len(cc.blockRandomness[cc.pendingBlocks[0].Hash]) != 0 ||
 			cc.pendingBlocks[0].Position.Round == 0) {
diff --git a/core/consensus.go b/core/consensus.go
index bf49a72..cc4b9e2 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -988,7 +988,7 @@ func (con *Consensus) ProcessBlockRandomnessResult(
 			"randomness", hex.EncodeToString(rand.Randomness))
 		con.network.BroadcastRandomnessResult(rand)
 	}
-	return nil
+	return con.deliverFinalizedBlocks()
 }
 
 // preProcessBlock performs Byzantine Agreement on the block.
@@ -1042,6 +1042,28 @@ func (con *Consensus) deliverBlock(b *types.Block) {
 	}
 }
 
+// deliverFinalizedBlocks extracts and delivers finalized blocks to application
+// layer.
+func (con *Consensus) deliverFinalizedBlocks() error {
+	con.lock.Lock()
+	defer con.lock.Unlock()
+	return con.deliverFinalizedBlocksWithoutLock()
+}
+
+func (con *Consensus) deliverFinalizedBlocksWithoutLock() (err error) {
+	deliveredBlocks := con.ccModule.extractBlocks()
+	con.logger.Debug("Last blocks in compaction chain",
+		"delivered", con.ccModule.lastDeliveredBlock(),
+		"pending", con.ccModule.lastPendingBlock())
+	for _, b := range deliveredBlocks {
+		con.deliverBlock(b)
+	}
+	if err = con.lattice.PurgeBlocks(deliveredBlocks); err != nil {
+		return
+	}
+	return
+}
+
 // processBlock is the entry point to submit one block to a Consensus instance.
 func (con *Consensus) processBlock(block *types.Block) (err error) {
 	if err = con.db.PutBlock(*block); err != nil && err != db.ErrBlockExists {
@@ -1075,14 +1097,7 @@ func (con *Consensus) processBlock(block *types.Block) (err error) {
 		}
 		go con.event.NotifyTime(b.Finalization.Timestamp)
 	}
-	deliveredBlocks = con.ccModule.extractBlocks()
-	con.logger.Debug("Last blocks in compaction chain",
-		"delivered", con.ccModule.lastDeliveredBlock(),
-		"pending", con.ccModule.lastPendingBlock())
-	for _, b := range deliveredBlocks {
-		con.deliverBlock(b)
-	}
-	if err = con.lattice.PurgeBlocks(deliveredBlocks); err != nil {
+	if err = con.deliverFinalizedBlocksWithoutLock(); err != nil {
 		return
 	}
 	return
diff --git a/core/db/level-db.go b/core/db/level-db.go
index 3b5994b..75c3037 100644
--- a/core/db/level-db.go
+++ b/core/db/level-db.go
@@ -163,7 +163,7 @@ func (lvl *LevelDBBackedDB) PutCompactionChainTipInfo(
 	if err != nil {
 		return err
 	}
-	if info.Height >= height {
+	if info.Height+1 != height {
 		return ErrInvalidCompactionChainTipHeight
 	}
 	if err = lvl.db.Put(compactionChainTipInfoKey, marshaled, nil); err != nil {
diff --git a/core/db/level-db_test.go b/core/db/level-db_test.go
index cf56b87..df971ee 100644
--- a/core/db/level-db_test.go
+++ b/core/db/level-db_test.go
@@ -141,14 +141,19 @@ func (s *LevelDBTestSuite) TestCompactionChainTipInfo() {
 	}(dbName)
 	// Save some tip info.
 	hash := common.NewRandomHash()
-	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 123))
+	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 1))
 	// Get it back to check.
 	hashBack, height := dbInst.GetCompactionChainTipInfo()
 	s.Require().Equal(hash, hashBack)
-	s.Require().Equal(height, uint64(123))
+	s.Require().Equal(height, uint64(1))
 	// Unable to put compaction chain tip info with lower height.
-	err = dbInst.PutCompactionChainTipInfo(hash, 122)
+	err = dbInst.PutCompactionChainTipInfo(hash, 0)
 	s.Require().IsType(err, ErrInvalidCompactionChainTipHeight)
+	// Unable to put compaction chain tip info with height not incremental by 1.
+	err = dbInst.PutCompactionChainTipInfo(hash, 3)
+	s.Require().IsType(err, ErrInvalidCompactionChainTipHeight)
+	// It's OK to put compaction chain tip info with height incremental by 1.
+	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 2))
 }
 
 func (s *LevelDBTestSuite) TestDKGPrivateKey() {
diff --git a/core/db/memory.go b/core/db/memory.go
index 7393de9..4bc08e7 100644
--- a/core/db/memory.go
+++ b/core/db/memory.go
@@ -149,7 +149,7 @@ func (m *MemBackedDB) PutCompactionChainTipInfo(
 	blockHash common.Hash, height uint64) error {
 	m.compactionChainTipLock.Lock()
 	defer m.compactionChainTipLock.Unlock()
-	if m.compactionChainTipHeight >= height {
+	if m.compactionChainTipHeight+1 != height {
 		return ErrInvalidCompactionChainTipHeight
 	}
 	m.compactionChainTipHeight = height
diff --git a/core/db/memory_test.go b/core/db/memory_test.go
index 09f74bb..a1b5165 100644
--- a/core/db/memory_test.go
+++ b/core/db/memory_test.go
@@ -137,14 +137,19 @@ func (s *MemBackedDBTestSuite) TestCompactionChainTipInfo() {
 	s.Require().NotNil(dbInst)
 	// Save some tip info.
 	hash := common.NewRandomHash()
-	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 123))
+	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 1))
 	// Get it back to check.
 	hashBack, height := dbInst.GetCompactionChainTipInfo()
 	s.Require().Equal(hash, hashBack)
-	s.Require().Equal(height, uint64(123))
+	s.Require().Equal(height, uint64(1))
 	// Unable to put compaction chain tip info with lower height.
-	err = dbInst.PutCompactionChainTipInfo(hash, 122)
+	err = dbInst.PutCompactionChainTipInfo(hash, 0)
 	s.Require().IsType(err, ErrInvalidCompactionChainTipHeight)
+	// Unable to put compaction chain tip info with height not incremental by 1.
+	err = dbInst.PutCompactionChainTipInfo(hash, 3)
+	s.Require().IsType(err, ErrInvalidCompactionChainTipHeight)
+	// It's OK to put compaction chain tip info with height incremental by 1.
+	s.Require().NoError(dbInst.PutCompactionChainTipInfo(hash, 2))
 }
 
 func (s *MemBackedDBTestSuite) TestDKGPrivateKey() {