Merge pull request #1 from youchainhq/develop

timestampfix

consensus/consensus.go

@@ -128,6 +128,10 @@ type Engine interface {
 
 	// Stop the engine
 	Stop() error
+
+	Pause() error
+
+	Resume() error
 }
 
 type Ucon interface {

consensus/errors.go

@@ -27,6 +27,9 @@ var (
 	// that is known, but the state of which is not available.
 	ErrPrunedAncestor = errors.New("pruned ancestor")
 
+	//new block should not older than parent
+	ErrOlderBlockTime    = errors.New("timestamp older than parent")
+
 	// ErrFutureBlock is returned when a block's timestamp is in the future according
 	// to the current node.
 	ErrFutureBlock = errors.New("block in the future")

consensus/solo/consensus.go

@@ -176,6 +176,14 @@ func (s *Solo) StartMining(chain consensus.ChainReader, inserter consensus.MineI
 	return nil
 }
 
+func (s *Solo) Pause() error {
+	return nil
+}
+
+func (s *Solo) Resume() error {
+	return nil
+}
+
 func (s *Solo) Stop() error {
 	//do nothing
 	return nil

consensus/ucon/consensus.go

@@ -138,6 +138,10 @@ func (s *Server) verifyCascadingFields(chain consensus.ChainReader, header *type
 		return consensus.ErrUnknownAncestor
 	}
 
+	if header.Time <= parentHeader.Time {
+		return consensus.ErrOlderBlockTime
+	}
+
 	local := chain.GetHeaderByNumber(header.Number.Uint64())
 	if local != nil && header.Hash() != local.Hash() {
 		logging.Error("Exist canonical", "localHash", local.Hash().String(), "headerHash", header.Hash().String())

consensus/ucon/timer_mgr.go

@@ -281,6 +281,18 @@ func (tm *TimerManager) Stop() {
 	close(tm.quitChan)
 }
 
+func (tm *TimerManager) Pause() {
+	tm.lock.Lock()
+	defer tm.lock.Unlock()
+
+	if tm.timeoutTimer != nil {
+		tm.timeoutTimer.Stop()
+	}
+	if tm.stepTimer != nil {
+		tm.stepTimer.Stop()
+	}
+}
+
 func (tm *TimerManager) NewAddressMsg(ev ReceivedMsgEvent, validatorKind params.ValidatorKind, chamberNum, houseNum uint64) bool {
 	tm.lock.Lock()
 	defer tm.lock.Unlock()

consensus/ucon/ucon.go

@@ -217,6 +217,7 @@ func (s *Server) StartMining(chain consensus.ChainReader, inserter consensus.Min
 // Stop the engine
 func (s *Server) Stop() error {
 	if !atomic.CompareAndSwapInt32(&s.alreadyStarted, 1, 0) {
+		logging.Error("Ucon already Stop")
 		return nil
 	}
 	logging.Info("Ucon Stop")
@@ -232,6 +233,44 @@ func (s *Server) Stop() error {
 	return nil
 }
 
+func (s *Server) Pause() error {
+	if !atomic.CompareAndSwapInt32(&s.alreadyStarted, 1, 0) {
+		logging.Error("Ucon already Stop")
+		return nil
+	}
+
+	s.timer.Pause()
+
+	logging.Info("consensus pause succeed")
+	return nil
+}
+
+func (s *Server) Resume() error {
+	logging.Info("try to resume")
+	if s.rawSk == nil {
+		return consensus.ErrValKeyNotSet
+	}
+	if s.blsSk == nil {
+		return consensus.ErrBlsKeyNotSet
+	}
+	s.startMiningLock.Lock()
+	defer s.startMiningLock.Unlock()
+	if s.isMining() {
+		logging.Info("Ucon Already Started")
+		return nil
+	}
+
+	err := s.StartNewRound(true)
+	if err != nil {
+		logging.Info("start new round error")
+		return err
+	}
+
+	atomic.StoreInt32(&s.alreadyStarted, 1)
+	logging.Info("consensus resume succeed")
+	return nil
+}
+
 func (s *Server) StartNewRound(newRound bool) error {
 	err := s.clearData(newRound)
 	if err != nil {
@@ -518,8 +557,11 @@ func (s *Server) updateBlockHeader(ev UpdateExistedHeaderEvent) {
 //HandleMsg handles related consensus messages or
 // fallback to default procotol manager's handler
 func (s *Server) HandleMsg(data []byte, receivedAt time.Time) error {
-	err := s.msgHandler.HandleMsg(data, receivedAt)
-	return err
+	if s.isMining() {
+		err := s.msgHandler.HandleMsg(data, receivedAt)
+		return err
+	}
+	return nil
 }
 
 func (s *Server) processTimeout(round *big.Int, roundIndex uint32) {

consensus/ucon/vote_bls.go

@@ -209,9 +209,10 @@ type VoteBLSMgr struct {
 
 func NewVoteBLSMgr(rawSk *ecdsa.PrivateKey, blsSk bls.SecretKey) *VoteBLSMgr {
 	vb := &VoteBLSMgr{
-		rawSk:  rawSk,
-		blsSk:  blsSk,
-		blsMgr: bls.NewBlsManager(),
+		rawSk:     rawSk,
+		blsSk:     blsSk,
+		blsMgr:    bls.NewBlsManager(),
+		currRound: new(big.Int),
 	}
 	vb.Verifier = NewBlsVerifier(vb.blsMgr)
 	if nil != rawSk {

consensus/ucon/vote_cache.go

@@ -90,10 +90,7 @@ func NewVoteDB(db youdb.Database, rawSk *ecdsa.PrivateKey) *VoteDB {
 }
 
 func (v *VoteDB) Stop() {
-	v.lock.Lock()
-	defer v.lock.Unlock()
 
-	v.db.Close()
 }
 
 func (v *VoteDB) UpdateContext(round *big.Int, roundIndex uint32) {

miner/miner.go

@@ -24,6 +24,7 @@ import (
 	"github.com/youchainhq/go-youchain/core/state"
 	"github.com/youchainhq/go-youchain/core/types"
 	"github.com/youchainhq/go-youchain/event"
+	"github.com/youchainhq/go-youchain/logging"
 	"github.com/youchainhq/go-youchain/node"
 	"github.com/youchainhq/go-youchain/params"
 	"github.com/youchainhq/go-youchain/you/downloader"
@@ -72,13 +73,19 @@ func (m *Miner) update() {
 		select {
 		case ev := <-events.Chan():
 			if ev == nil {
+				logging.Info("events.Chan quit")
 				return
 			}
 			switch ev.Data.(type) {
 			case downloader.StartEvent:
 				log.Info("miner recv StartEvent")
 				atomic.StoreInt32(&m.canStart, 0)
 				if m.worker.isRunning() {
+					log.Error("miner stop consensus engine")
+					if err := m.engine.Pause(); err != nil {
+						log.Error("miner pause consensus engine", "error", err)
+					}
+
 					m.Stop()
 					atomic.StoreInt32(&m.shouldStart, 1)
 					log.Info("mining aborted due to sync")
@@ -90,10 +97,14 @@ func (m *Miner) update() {
 				atomic.StoreInt32(&m.canStart, 1)
 				atomic.StoreInt32(&m.shouldStart, 0)
 				if shouldStart {
+
+					log.Info("miner restart consensus engine")
+					if err := m.engine.Resume(); err != nil {
+						log.Error("miner resume consensus engine", "error", err)
+					}
+
 					m.Start()
 				}
-				// stop immediately and ignore all further pending events
-				return
 			}
 		case <-m.quit:
 			return

miner/worker.go

@@ -270,7 +270,7 @@ func (w *worker) commitNewWork(interrupt *int32) {
 
 	//quick fast
 	if !w.isRunning() {
-		log.Info("worker is not running")
+		log.Warn("worker is not running")
 		return
 	}
 
@@ -279,6 +279,12 @@ func (w *worker) commitNewWork(interrupt *int32) {
 	coinbase.SetBytes(w.engine.GetValMainAddress().Bytes())
 
 	parent := w.chain.CurrentBlock()
+
+	if uint64(time.Now().Unix()) <= parent.Time() {
+		log.Warn("local time is far behind")
+		return
+	}
+
 	num := parent.Number()
 	header := &types.Header{
 		ParentHash: parent.Hash(),

you/backend.go

@@ -63,6 +63,8 @@ type YouChain struct {
 	//DB
 	chainDb youdb.Database
 
+	voteDb youdb.Database
+
 	accountManager *accounts.Manager
 
 	BloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
@@ -120,6 +122,7 @@ func New(config *Config, nodeConfig *node.Config) (*YouChain, error) {
 		nodeConfig:    nodeConfig,
 		quit:          make(chan bool),
 		chainDb:       chainDb,
+		voteDb:        voteDb,
 		engine:        engine,
 		eventMux:      eventMux,
 		BloomRequests: make(chan chan *bloombits.Retrieval),
@@ -342,6 +345,16 @@ func (you *YouChain) StartMining() error {
 	return nil
 }
 
+func (you *YouChain) StopMining() error {
+	err := you.engine.Stop()
+	if err != nil {
+		logging.Error("stop mining", "err", err)
+	}
+
+	you.miner.Stop()
+	return nil
+}
+
 func (you *YouChain) Stop() error {
 	you.blockChain.Stop()
 
@@ -352,6 +365,7 @@ func (you *YouChain) Stop() error {
 	you.eventMux.Stop()
 
 	you.chainDb.Close()
+	you.voteDb.Close()
 	you.stakingMan.Stop()
 
 	close(you.quit)

you/miner_api.go

@@ -36,8 +36,8 @@ func (api *PrivateMinerApi) Start() error {
 	return api.c.StartMining()
 }
 
-func (api *PrivateMinerApi) Stop() {
-	api.c.miner.Stop()
+func (api *PrivateMinerApi) Stop() error {
+	return api.c.StopMining()
 }
 
 // Mining returns an indication if this node is currently mining.