core: consensus-timestamp: add sync (#219)

* core: consensus-timestamp: add sync

* core: consensus-timestamp: add config change

* fix go comment

* add config change test

* fixup: add error case handling

* fixup: round interleave

3 files changed, 118 insertions, 29 deletions

diff --git a/core/consensus-timestamp.go b/core/consensus-timestamp.go
index 9551328..dc6d2a8 100644
--- a/core/consensus-timestamp.go
+++ b/core/consensus-timestamp.go
@@ -29,8 +29,8 @@ type consensusTimestamp struct {
 	chainTimestamps []time.Time
 
 	// This part keeps configs for each round.
-	numChainsForRounds []uint32
-	numChainsRoundBase uint64
+	numChainsOfRounds []uint32
+	numChainsBase     uint64
 
 	// dMoment represents the genesis time.
 	dMoment time.Time
@@ -40,15 +40,22 @@ var (
 	// ErrTimestampNotIncrease would be reported if the timestamp is not strickly
 	// increasing on the same chain.
 	ErrTimestampNotIncrease = errors.New("timestamp is not increasing")
+	// ErrNoRoundConfig for no round config found.
+	ErrNoRoundConfig = errors.New("no round config found")
 )
 
 // newConsensusTimestamp creates timestamper object.
 func newConsensusTimestamp(
-	dMoment time.Time, numChains uint32) *consensusTimestamp {
+	dMoment time.Time, round uint64, numChains uint32) *consensusTimestamp {
+	ts := make([]time.Time, 0, numChains)
+	for i := uint32(0); i < numChains; i++ {
+		ts = append(ts, dMoment)
+	}
 	return &consensusTimestamp{
-		numChainsForRounds: []uint32{numChains},
-		numChainsRoundBase: uint64(0),
-		dMoment:            dMoment,
+		numChainsOfRounds: []uint32{numChains},
+		numChainsBase:     round,
+		dMoment:           dMoment,
+		chainTimestamps:   ts,
 	}
 }
 
@@ -57,29 +64,49 @@ func newConsensusTimestamp(
 func (ct *consensusTimestamp) appendConfig(
 	round uint64, config *types.Config) error {
 
-	if round != uint64(len(ct.numChainsForRounds))+ct.numChainsRoundBase {
+	if round != uint64(len(ct.numChainsOfRounds))+ct.numChainsBase {
 		return ErrRoundNotIncreasing
 	}
-	ct.numChainsForRounds = append(ct.numChainsForRounds, config.NumChains)
+	ct.numChainsOfRounds = append(ct.numChainsOfRounds, config.NumChains)
 	return nil
 }
 
-func (ct *consensusTimestamp) getNumChains(round uint64) uint32 {
-	roundIndex := round - ct.numChainsRoundBase
-	return ct.numChainsForRounds[roundIndex]
+func (ct *consensusTimestamp) resizeChainTimetamps(numChain uint32) {
+	l := uint32(len(ct.chainTimestamps))
+	if numChain > l {
+		for i := l; i < numChain; i++ {
+			ct.chainTimestamps = append(ct.chainTimestamps, ct.dMoment)
+		}
+	} else if numChain < l {
+		ct.chainTimestamps = ct.chainTimestamps[:numChain]
+	}
 }
 
 // ProcessBlocks is the entry function.
 func (ct *consensusTimestamp) processBlocks(blocks []*types.Block) (err error) {
 	for _, block := range blocks {
-		numChains := ct.getNumChains(block.Position.Round)
-		// Fulfill empty time slots with d-moment. This part also means
-		// each time we increasing number of chains, we can't increase over
-		// 49% of previous number of chains.
-		for uint32(len(ct.chainTimestamps)) < numChains {
-			ct.chainTimestamps = append(ct.chainTimestamps, ct.dMoment)
+		// Rounds might interleave within rounds if no configuration change
+		// occurs. And it is limited to one round, that is, round r can only
+		// interleave with r-1 and r+1.
+		round := block.Position.Round
+		if ct.numChainsBase == round || ct.numChainsBase+1 == round {
+			// Normal case, no need to modify chainTimestamps.
+		} else if ct.numChainsBase+2 == round {
+			if len(ct.numChainsOfRounds) < 2 {
+				return ErrNoRoundConfig
+			}
+			if ct.numChainsOfRounds[0] > ct.numChainsOfRounds[1] {
+				ct.resizeChainTimetamps(ct.numChainsOfRounds[0])
+			} else {
+				ct.resizeChainTimetamps(ct.numChainsOfRounds[1])
+			}
+			ct.numChainsBase++
+			ct.numChainsOfRounds = ct.numChainsOfRounds[1:]
+		} else {
+			// Error if round < base or round > base + 2.
+			return ErrInvalidRoundID
 		}
-		ts := ct.chainTimestamps[:numChains]
+		ts := ct.chainTimestamps[:ct.numChainsOfRounds[round-ct.numChainsBase]]
 		if block.Finalization.Timestamp, err = getMedianTime(ts); err != nil {
 			return
 		}
@@ -87,12 +114,16 @@ func (ct *consensusTimestamp) processBlocks(blocks []*types.Block) (err error) {
 			return ErrTimestampNotIncrease
 		}
 		ct.chainTimestamps[block.Position.ChainID] = block.Timestamp
-		// Purge configs for older rounds, rounds of blocks from total ordering
-		// would increase.
-		if block.Position.Round > ct.numChainsRoundBase {
-			ct.numChainsRoundBase++
-			ct.numChainsForRounds = ct.numChainsForRounds[1:]
-		}
 	}
 	return
 }
+
+func (ct *consensusTimestamp) isSynced() bool {
+	numChain := ct.numChainsOfRounds[0]
+	for i := uint32(0); i < numChain; i++ {
+		if ct.chainTimestamps[i].Equal(ct.dMoment) {
+			return false
+		}
+	}
+	return true
+}
diff --git a/core/consensus-timestamp_test.go b/core/consensus-timestamp_test.go
index 49002aa..0f2c526 100644
--- a/core/consensus-timestamp_test.go
+++ b/core/consensus-timestamp_test.go
@@ -94,7 +94,7 @@ func (s *ConsensusTimestampTest) TestTimestampPartition() {
 	chainNum := 19
 	sigma := 100 * time.Millisecond
 	totalTimestamps := make([]time.Time, 0)
-	ct := newConsensusTimestamp(time.Time{}, uint32(chainNum))
+	ct := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
 	totalBlockNum := 0
 	for _, blockNum := range blockNums {
 		totalBlockNum += blockNum
@@ -110,7 +110,7 @@ func (s *ConsensusTimestampTest) TestTimestampPartition() {
 		totalChain = append(totalChain, chain...)
 		totalTimestamps = append(totalTimestamps, timestamps...)
 	}
-	ct2 := newConsensusTimestamp(time.Time{}, uint32(chainNum))
+	ct2 := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
 	err := ct2.processBlocks(totalChain)
 	s.Require().NoError(err)
 	timestamps2 := s.extractTimestamps(totalChain)
@@ -120,7 +120,7 @@ func (s *ConsensusTimestampTest) TestTimestampPartition() {
 func (s *ConsensusTimestampTest) TestTimestampIncrease() {
 	chainNum := 19
 	sigma := 100 * time.Millisecond
-	ct := newConsensusTimestamp(time.Time{}, uint32(chainNum))
+	ct := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
 	chain := s.generateBlocksWithTimestamp(1000, chainNum, time.Second, sigma)
 	err := ct.processBlocks(chain)
 	s.Require().NoError(err)
@@ -129,13 +129,71 @@ func (s *ConsensusTimestampTest) TestTimestampIncrease() {
 		s.False(timestamps[i].Before(timestamps[i-1]))
 	}
 	// Test if the processBlocks is stable.
-	ct2 := newConsensusTimestamp(time.Time{}, uint32(chainNum))
+	ct2 := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
 	ct2.processBlocks(chain)
 	s.Require().NoError(err)
 	timestamps2 := s.extractTimestamps(chain)
 	s.Equal(timestamps, timestamps2)
 }
 
+func (s *ConsensusTimestampTest) TestTimestampConfigChange() {
+	chainNum := 19
+	sigma := 100 * time.Millisecond
+	ct := newConsensusTimestamp(time.Time{}, 20, uint32(chainNum))
+	chain := s.generateBlocksWithTimestamp(1000, chainNum, time.Second, sigma)
+	blocks := make([]*types.Block, 0, 1000)
+	ct.appendConfig(21, &types.Config{NumChains: uint32(16)})
+	ct.appendConfig(22, &types.Config{NumChains: uint32(19)})
+	// Blocks 0 to 299 is in round 20, blocks 300 to 599 is in round 21 and ignore
+	// blocks which ChainID is 16 to 18, blocks 600 to 999 is in round 22.
+	for i := 0; i < 1000; i++ {
+		add := true
+		if i < 300 {
+			chain[i].Position.Round = 20
+		} else if i < 600 {
+			chain[i].Position.Round = 21
+			add = chain[i].Position.ChainID < 16
+		} else {
+			chain[i].Position.Round = 22
+		}
+		if add {
+			blocks = append(blocks, chain[i])
+		}
+	}
+	err := ct.processBlocks(blocks)
+	s.Require().NoError(err)
+}
+
+func (s *ConsensusTimestampTest) TestRoundInterleave() {
+	chainNum := 9
+	sigma := 100 * time.Millisecond
+	ct := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
+	ct.appendConfig(1, &types.Config{NumChains: uint32(chainNum)})
+	chain := s.generateBlocksWithTimestamp(100, chainNum, time.Second, sigma)
+	for i := 50; i < 100; i++ {
+		chain[i].Position.Round = 1
+	}
+	chain[48].Position.Round = 1
+	chain[49].Position.Round = 1
+	chain[50].Position.Round = 0
+	chain[51].Position.Round = 0
+	err := ct.processBlocks(chain)
+	s.Require().NoError(err)
+}
+
+func (s *ConsensusTimestampTest) TestTimestampSync() {
+	chainNum := 19
+	sigma := 100 * time.Millisecond
+	ct := newConsensusTimestamp(time.Time{}, 0, uint32(chainNum))
+	chain := s.generateBlocksWithTimestamp(100, chainNum, time.Second, sigma)
+	err := ct.processBlocks(chain[:chainNum-1])
+	s.Require().NoError(err)
+	s.Require().False(ct.isSynced())
+	err = ct.processBlocks(chain[chainNum-1:])
+	s.Require().NoError(err)
+	s.Require().True(ct.isSynced())
+}
+
 func TestConsensusTimestamp(t *testing.T) {
 	suite.Run(t, new(ConsensusTimestampTest))
 }
diff --git a/core/lattice.go b/core/lattice.go
index a0028a7..b481869 100644
--- a/core/lattice.go
+++ b/core/lattice.go
@@ -58,7 +58,7 @@ func NewLattice(
 		pool:       newBlockPool(cfg.NumChains),
 		data:       newLatticeData(db, dataConfig),
 		toModule:   newTotalOrdering(toConfig),
-		ctModule:   newConsensusTimestamp(dMoment, cfg.NumChains),
+		ctModule:   newConsensusTimestamp(dMoment, 0, cfg.NumChains),
 	}
 	return
 }