core: syncer: fix force sync (#556)

* core: syncer: fix force sync

* Fix come logic

* core: syncer: fix

* fix log

4 files changed, 19 insertions, 5 deletions

diff --git a/core/syncer/consensus.go b/core/syncer/consensus.go
index f4681a2..7db836a 100644
--- a/core/syncer/consensus.go
+++ b/core/syncer/consensus.go
@@ -276,17 +276,21 @@ func (con *Consensus) buildAllEmptyBlocks() {
 }
 
 // ForceSync forces syncer to become synced.
-func (con *Consensus) ForceSync(skip bool) {
+func (con *Consensus) ForceSync(lastPos types.Position, skip bool) {
 	if con.syncedLastBlock != nil {
 		return
 	}
-	hash, _ := con.db.GetCompactionChainTipInfo()
-	var block types.Block
+	hash, height := con.db.GetCompactionChainTipInfo()
+	if height < lastPos.Height {
+		panic(fmt.Errorf("compaction chain not synced height %d, tip %d",
+			lastPos.Height, height))
+	} else if height > lastPos.Height {
+		skip = false
+	}
 	block, err := con.db.GetBlock(hash)
 	if err != nil {
 		panic(err)
 	}
-	con.logger.Info("Force Sync", "block", &block)
 	con.syncedLastBlock = &block
 	con.stopBuffering()
 	// We might call stopBuffering without calling assureBuffering.
@@ -298,6 +302,7 @@ func (con *Consensus) ForceSync(skip bool) {
 			})
 	}
 	con.syncedSkipNext = skip
+	con.logger.Info("Force Sync", "block", &block, "skip", skip)
 }
 
 // SyncBlocks syncs blocks from compaction chain, latest is true if the caller
diff --git a/core/syncer/watch-cat.go b/core/syncer/watch-cat.go
index d08bff9..f2e197e 100644
--- a/core/syncer/watch-cat.go
+++ b/core/syncer/watch-cat.go
@@ -37,6 +37,7 @@ type WatchCat struct {
 	timeout      time.Duration
 	configReader configReader
 	feed         chan types.Position
+	lastPosition types.Position
 	polling      time.Duration
 	ctx          context.Context
 	cancel       context.CancelFunc
@@ -69,6 +70,7 @@ func (wc *WatchCat) Feed(position types.Position) {
 // Start the WatchCat.
 func (wc *WatchCat) Start() {
 	wc.Stop()
+	wc.lastPosition = types.Position{}
 	wc.ctx, wc.cancel = context.WithCancel(context.Background())
 	go func() {
 		var lastPos types.Position
@@ -124,6 +126,7 @@ func (wc *WatchCat) Start() {
 				wc.logger.Error("Failed to get recovery votes", "height", lastPos.Height, "error", err)
 			} else if votes > threshold {
 				wc.logger.Info("Threshold for recovery reached!")
+				wc.lastPosition = lastPos
 				break ResetLoop
 			}
 			select {
@@ -146,3 +149,8 @@ func (wc *WatchCat) Stop() {
 func (wc *WatchCat) Meow() <-chan struct{} {
 	return wc.ctx.Done()
 }
+
+// LastPosition returns the last position for recovery.
+func (wc *WatchCat) LastPosition() types.Position {
+	return wc.lastPosition
+}
diff --git a/core/syncer/watch-cat_test.go b/core/syncer/watch-cat_test.go
index 103200d..b1f1bf8 100644
--- a/core/syncer/watch-cat_test.go
+++ b/core/syncer/watch-cat_test.go
@@ -116,6 +116,7 @@ func (s *WatchCatTestSuite) TestBasicUsage() {
 	default:
 		s.FailNow("expecting terminated")
 	}
+	s.Equal(pos, watchCat.LastPosition())
 }
 
 func TestWatchCat(t *testing.T) {
diff --git a/integration_test/consensus_test.go b/integration_test/consensus_test.go
index ae56af2..05d7c44 100644
--- a/integration_test/consensus_test.go
+++ b/integration_test/consensus_test.go
@@ -677,7 +677,7 @@ ReachStop:
 		s.Require().Equal(latestPos, pos)
 	}
 	for _, con := range syncerCon {
-		con.ForceSync(true)
+		con.ForceSync(latestPos, true)
 	}
 	for nID := range nodes {
 		con, err := syncerCon[nID].GetSyncedConsensus()