verify RenameData() versions to be consistent (#15649)

xl.meta gets written and never rolled back, however
we definitely need to validate the state that is
persisted on the disk, if there are inconsistencies

- more than write quorum we should return an error
  to the client

- if write quorum was achieved however there are
  inconsistent xl.meta's we should simply trigger
  an MRF on them

cmd/xl-storage.go

@@ -46,6 +46,7 @@ import (
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/console"
 	"github.com/yargevad/filepathx"
+	"github.com/zeebo/xxh3"
 )
 
 const (
@@ -2132,7 +2133,7 @@ func skipAccessChecks(volume string) (ok bool) {
 }
 
 // RenameData - rename source path to destination path atomically, metadata and data directory.
-func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) (err error) {
+func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) (sign uint64, err error) {
 	defer func() {
 		if err != nil && !contextCanceled(ctx) && !errors.Is(err, errFileNotFound) {
 			// Only log these errors if context is not yet canceled.
@@ -2150,34 +2151,34 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 
 	srcVolumeDir, err := s.getVolDir(srcVolume)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	dstVolumeDir, err := s.getVolDir(dstVolume)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	if !skipAccessChecks(srcVolume) {
 		// Stat a volume entry.
 		if err = Access(srcVolumeDir); err != nil {
 			if osIsNotExist(err) {
-				return errVolumeNotFound
+				return 0, errVolumeNotFound
 			} else if isSysErrIO(err) {
-				return errFaultyDisk
+				return 0, errFaultyDisk
 			}
-			return err
+			return 0, err
 		}
 	}
 
 	if !skipAccessChecks(dstVolume) {
 		if err = Access(dstVolumeDir); err != nil {
 			if osIsNotExist(err) {
-				return errVolumeNotFound
+				return 0, errVolumeNotFound
 			} else if isSysErrIO(err) {
-				return errFaultyDisk
+				return 0, errFaultyDisk
 			}
-			return err
+			return 0, err
 		}
 	}
 
@@ -2199,11 +2200,11 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 	}
 
 	if err = checkPathLength(srcFilePath); err != nil {
-		return err
+		return 0, err
 	}
 
 	if err = checkPathLength(dstFilePath); err != nil {
-		return err
+		return 0, err
 	}
 
 	dstBuf, err := xioutil.ReadFile(dstFilePath)
@@ -2215,20 +2216,20 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 		if isSysErrNotDir(err) && runtime.GOOS != globalWindowsOSName {
 			// NOTE: On windows the error happens at
 			// next line and returns appropriate error.
-			return errFileAccessDenied
+			return 0, errFileAccessDenied
 		}
 		if !osIsNotExist(err) {
-			return osErrToFileErr(err)
+			return 0, osErrToFileErr(err)
 		}
 		// errFileNotFound comes here.
 		err = s.renameLegacyMetadata(dstVolumeDir, dstPath)
 		if err != nil && err != errFileNotFound {
-			return err
+			return 0, err
 		}
 		if err == nil {
 			dstBuf, err = xioutil.ReadFile(dstFilePath)
 			if err != nil && !osIsNotExist(err) {
-				return osErrToFileErr(err)
+				return 0, osErrToFileErr(err)
 			}
 		}
 	}
@@ -2272,7 +2273,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 			currentDataPath := pathJoin(dstVolumeDir, dstPath)
 			entries, err := readDirN(currentDataPath, 1)
 			if err != nil && err != errFileNotFound {
-				return osErrToFileErr(err)
+				return 0, osErrToFileErr(err)
 			}
 			for _, entry := range entries {
 				if entry == xlStorageFormatFile || strings.HasSuffix(entry, slashSeparator) {
@@ -2292,14 +2293,14 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 		currentDataPath := pathJoin(dstVolumeDir, dstPath)
 		entries, err := readDir(currentDataPath)
 		if err != nil {
-			return osErrToFileErr(err)
+			return 0, osErrToFileErr(err)
 		}
 
 		// legacy data dir means its old content, honor system umask.
 		if err = mkdirAll(legacyDataPath, 0o777); err != nil {
 			// any failed mkdir-calls delete them.
 			s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
-			return osErrToFileErr(err)
+			return 0, osErrToFileErr(err)
 		}
 
 		for _, entry := range entries {
@@ -2312,7 +2313,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 				// Any failed rename calls un-roll previous transaction.
 				s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 
-				return osErrToFileErr(err)
+				return 0, osErrToFileErr(err)
 			}
 		}
 	}
@@ -2357,9 +2358,15 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 			// Any failed rename calls un-roll previous transaction.
 			s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 		}
-		return err
+		return 0, err
 	}
 
+	var sbuf bytes.Buffer
+	for _, ver := range xlMeta.versions {
+		sbuf.Write(ver.header.Signature[:])
+	}
+	sign = xxh3.Hash(sbuf.Bytes())
+
 	dstBuf, err = xlMeta.AppendTo(metaDataPoolGet())
 	defer metaDataPoolPut(dstBuf)
 	if err != nil {
@@ -2368,7 +2375,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 			// Any failed rename calls un-roll previous transaction.
 			s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 		}
-		return errFileCorrupt
+		return 0, errFileCorrupt
 	}
 
 	if srcDataPath != "" {
@@ -2377,7 +2384,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 				// Any failed rename calls un-roll previous transaction.
 				s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 			}
-			return osErrToFileErr(err)
+			return 0, osErrToFileErr(err)
 		}
 		diskHealthCheckOK(ctx, err)
 
@@ -2396,7 +2403,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 					s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 				}
 				s.deleteFile(dstVolumeDir, dstDataPath, false, false)
-				return osErrToFileErr(err)
+				return 0, osErrToFileErr(err)
 			}
 		}
 
@@ -2407,7 +2414,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 				s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 			}
 			s.deleteFile(dstVolumeDir, dstFilePath, false, false)
-			return osErrToFileErr(err)
+			return 0, osErrToFileErr(err)
 		}
 
 		// additionally only purge older data at the end of the transaction of new data-dir
@@ -2424,7 +2431,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 				s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
 			}
 			s.deleteFile(dstVolumeDir, dstFilePath, false, false)
-			return err
+			return 0, err
 		}
 	}
 
@@ -2433,7 +2440,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 	// ideally all transaction should be complete.
 
 	Remove(pathutil.Dir(srcFilePath))
-	return nil
+	return sign, nil
 }
 
 // RenameFile - rename source path to destination path atomically.