Merge remote-tracking branch 'couchbase/unstable' into HEAD

http: //ci2i-unstable.northscale.in/gsi-04.08.2022-14.25.pass.html
Change-Id: I59b428ac2c206f59a41552580d44395d5642470c

secondary/indexer/storage_manager.go

@@ -984,9 +984,9 @@ func (sm *storageMgr) rollbackToSnapshot(idxInstId common.IndexInstId,
 			logging.Infof("StorageMgr::handleRollback Rollback Index: %v "+
 				"PartitionId: %v SliceId: %v To Snapshot %v ", idxInstId, partnId,
 				slice.Id(), snapInfo)
-			restartTs = snapInfo.Timestamp()
+			restartTs = snapInfo.Timestamp().Copy()
 			if markAsUsed {
-				slice.SetLastRollbackTs(restartTs)
+				slice.SetLastRollbackTs(restartTs.Copy())
 			}
 		} else {
 			//send error response back

secondary/indexer/stream_state.go

@@ -1459,6 +1459,11 @@ func (ss *StreamState) getNextStabilityTS(streamId common.StreamId,
 	//generate new stability timestamp
 	tsVbuuid := ss.streamKeyspaceIdHWTMap[streamId][keyspaceId].Copy()
 
+	//Explicitly set the snapAligned flag to false as the initial state.
+	//HWT can be set from an earlier restartTs which is snap aligned
+	//and the flag needs to be reset for new timestamp.
+	tsVbuuid.SetSnapAligned(false)
+
 	tsElem := &TsListElem{
 		ts: tsVbuuid,
 	}
@@ -1479,6 +1484,8 @@ func (ss *StreamState) getNextStabilityTS(streamId common.StreamId,
 
 	if tsElem.ts.CheckSnapAligned() {
 		tsElem.ts.SetSnapAligned(true)
+	} else {
+		tsElem.ts.SetSnapAligned(false)
 	}
 
 	return tsElem
@@ -1699,7 +1706,10 @@ func (ss *StreamState) computeTsChangeVec(streamId common.StreamId,
 	if enableOSO {
 		if ts.CheckSnapAligned() {
 			ts.SetSnapAligned(true)
+		} else {
+			ts.SetSnapAligned(false)
 		}
+
 	}
 
 	return changeVec, noChange, countVec

secondary/indexer/timekeeper.go

@@ -3310,6 +3310,8 @@ func (tk *timekeeper) mayBeMakeSnapAligned(streamId common.StreamId,
 
 	if flushTs.CheckSnapAligned() {
 		flushTs.SetSnapAligned(true)
+	} else {
+		flushTs.SetSnapAligned(false)
 	}
 
 }