feat: detect corrupted checkpoint when checkpoint seqNo bigger then l…

…atest
Trendyol · Jul 20, 2024 · f766145 · f766145
1 parent c06c48a
commit f766145
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 14 deletions.
diff --git a/couchbase/observer.go b/couchbase/observer.go
@@ -89,7 +89,7 @@ func (so *observer) persistSeqNoChangedListener(persistSeqNo models.PersistSeqNo
 			so.persistSeqNo.Store(persistSeqNo.VbID, persistSeqNo.SeqNo)
 		}
 	} else {
-		logger.Log.Trace("persistSeqNo: %v on vbId: %v", persistSeqNo.SeqNo, persistSeqNo.VbID)
+		logger.Log.Trace("persistSeqNo: %v on vbID: %v", persistSeqNo.SeqNo, persistSeqNo.VbID)
 	}
 }
 

diff --git a/couchbase/rollback_mitigation.go b/couchbase/rollback_mitigation.go
@@ -119,7 +119,7 @@ func (r *rollbackMitigation) observeVbID(
 		Deadline:   time.Now().Add(time.Second * 5),
 	}, callback)
 	if err != nil {
-		logger.Log.Error("observeVBID error for vbId: %v, replica:%v, vbUUID: %v, err: %v", vbID, replica, vbUUID, err)
+		logger.Log.Error("observeVBID error for vbID: %v, replica:%v, vbUUID: %v, err: %v", vbID, replica, vbUUID, err)
 		callback(nil, err)
 	}
 }
@@ -170,15 +170,15 @@ func (r *rollbackMitigation) markAbsentInstances() error { //nolint:unused
 			serverIndex, err := r.configSnapshot.VbucketToServer(vbID, uint32(idx))
 			if err != nil {
 				if errors.Is(err, gocbcore.ErrInvalidReplica) {
-					logger.Log.Debug("invalid replica of vbId: %v, replica: %v, err: %v", vbID, idx, err)
+					logger.Log.Debug("invalid replica of vbID: %v, replica: %v, err: %v", vbID, idx, err)
 					replica.SetAbsent()
 				} else {
 					outerError = err
 					return false
 				}
 			} else {
 				if serverIndex < 0 {
-					logger.Log.Debug("invalid server index of vbId: %v, replica: %v, serverIndex: %v", vbID, idx, serverIndex)
+					logger.Log.Debug("invalid server index of vbID: %v, replica: %v, serverIndex: %v", vbID, idx, serverIndex)
 					replica.SetAbsent()
 				}
 			}

diff --git a/stream/checkpoint.go b/stream/checkpoint.go
@@ -1,6 +1,7 @@
 package stream
 
 import (
+	"errors"
 	"sync"
 	"time"
 
@@ -109,6 +110,7 @@ func (s *checkpoint) Save() {
 	}
 }
 
+//nolint:funlen
 func (s *checkpoint) Load() (*wrapper.ConcurrentSwissMap[uint16, *models.Offset], *wrapper.ConcurrentSwissMap[uint16, bool], bool) {
 	s.loadLock.Lock()
 	defer s.loadLock.Unlock()
@@ -121,19 +123,19 @@ func (s *checkpoint) Load() (*wrapper.ConcurrentSwissMap[uint16, *models.Offset]
 		panic(err)
 	}
 
+	seqNoMap, err := s.client.GetVBucketSeqNos()
+	if err != nil {
+		logger.Log.Error("error while getting vBucket seqNos, err: %v", err)
+		panic(err)
+	}
+
 	offsets := wrapper.CreateConcurrentSwissMap[uint16, *models.Offset](1024)
 	dirtyOffsets := wrapper.CreateConcurrentSwissMap[uint16, bool](1024)
 	anyDirtyOffset := false
 
 	if !exist && s.config.Checkpoint.AutoReset == CheckpointAutoResetTypeLatest {
 		logger.Log.Debug("no checkpoint found, auto reset checkpoint to latest")
 
-		seqNoMap, err := s.client.GetVBucketSeqNos()
-		if err != nil {
-			logger.Log.Error("error while getting vBucket seqNos, err: %v", err)
-			panic(err)
-		}
-
 		dump.Range(func(vbID uint16, doc *models.CheckpointDocument) bool {
 			currentSeqNo, _ := seqNoMap.Load(vbID)
 
@@ -158,6 +160,16 @@ func (s *checkpoint) Load() (*wrapper.ConcurrentSwissMap[uint16, *models.Offset]
 	}
 
 	dump.Range(func(vbID uint16, doc *models.CheckpointDocument) bool {
+		latestSeqNo, _ := seqNoMap.Load(vbID)
+		if doc.Checkpoint.SeqNo > latestSeqNo {
+			err := errors.New("checkpoint seqNo bigger then vBucket latest seqNo")
+			logger.Log.Error(
+				"error while loading checkpoint, vbID: %v, checkpoint seqNo: %v, latest seqNo: %v, err: %v",
+				vbID, doc.Checkpoint.SeqNo, latestSeqNo, err,
+			)
+			panic(err)
+		}
+
 		offsets.Store(vbID, &models.Offset{
 			SnapshotMarker: &models.SnapshotMarker{
 				StartSeqNo: doc.Checkpoint.Snapshot.StartSeqNo,

diff --git a/stream/stream.go b/stream/stream.go
@@ -78,7 +78,7 @@ func (s *stream) setOffset(vbID uint16, offset *models.Offset, dirty bool) {
 		s.offsets.Store(vbID, offset)
 		s.dirtyOffsets.Store(vbID, dirty)
 	} else {
-		logger.Log.Warn("vbId=%v not belong our vbId range", vbID)
+		logger.Log.Warn("vbID: %v not belong our vbId range", vbID)
 	}
 }
 
@@ -164,14 +164,14 @@ func (s *stream) listenEnd() {
 	for endContext := range s.observer.ListenEnd() {
 		if !s.closeWithCancel && endContext.Err != nil {
 			if !errors.Is(endContext.Err, gocbcore.ErrDCPStreamClosed) {
-				logger.Log.Error("end stream vbId: %v got error: %v", endContext.Event.VbID, endContext.Err)
+				logger.Log.Error("end stream vbID: %v got error: %v", endContext.Event.VbID, endContext.Err)
 			} else {
-				logger.Log.Debug("end stream vbId: %v got error: %v", endContext.Event.VbID, endContext.Err)
+				logger.Log.Debug("end stream vbID: %v got error: %v", endContext.Event.VbID, endContext.Err)
 			}
 		}
 
 		if endContext.Err == nil {
-			logger.Log.Debug("end stream vbId: %v", endContext.Event.VbID)
+			logger.Log.Debug("end stream vbID: %v", endContext.Event.VbID)
 		}
 
 		if !s.closeWithCancel && endContext.Err != nil &&