fix: block snapshot events when persisted seqNo lower then snapshots …

…startSeqNo with rollback mitigation enabled

couchbase/observer.go

@@ -165,6 +165,10 @@ func (so *observer) sendOrSkip(args models.ListenerArgs) {
 }
 
 func (so *observer) SnapshotMarker(event models.DcpSnapshotMarker) {
+	if !so.canForward(event.VbID, event.StartSeqNo) {
+		return
+	}
+
 	so.currentSnapshots.Store(event.VbID, &models.SnapshotMarker{
 		StartSeqNo: event.StartSeqNo,
 		EndSeqNo:   event.EndSeqNo,

stream/stream.go

@@ -79,8 +79,21 @@ type stream struct {
 
 func (s *stream) setOffset(vbID uint16, offset *models.Offset, dirty bool) {
 	if _, ok := s.vbIds.Load(vbID); ok {
+		if current, ok := s.offsets.Load(vbID); ok && current.SeqNo > offset.SeqNo {
+			return
+		}
 		s.offsets.Store(vbID, offset)
-		s.dirtyOffsets.Store(vbID, dirty)
+		if !dirty {
+			return
+		}
+
+		s.dirtyOffsets.StoreIf(vbID, func(p bool, f bool) (v bool, s bool) {
+			if !f || (f && !p) {
+				return true, true
+			}
+
+			return p, false
+		})
 	} else {
 		logger.Log.Warn("vbID: %v not belong our vbId range", vbID)
 	}

wrapper/concurrent_swiss_map.go

@@ -35,6 +35,10 @@ func (m *ConcurrentSwissMap[K, V]) Store(key K, value V) {
 	m.m.Store(key, value)
 }
 
+func (m *ConcurrentSwissMap[K, V]) StoreIf(key K, conditionFn func(previousVale V, previousFound bool) (value V, set bool)) {
+	m.m.SetIf(key, conditionFn)
+}
+
 func (m *ConcurrentSwissMap[K, V]) Count() int {
 	return m.m.Count()
 }