Return NumVersions in quorum when available (#19766)

Similar to https://github.com/minio/minio/pull/17925

cmd/erasure-metadata-utils.go

@@ -26,6 +26,20 @@ import (
 	"github.com/minio/pkg/v2/sync/errgroup"
 )
 
+// counterMap type adds GetValueWithQuorum method to a map[T]int used to count occurrences of values of type T.
+type counterMap[T comparable] map[T]int
+
+// GetValueWithQuorum returns the first key which occurs >= quorum number of times.
+func (c counterMap[T]) GetValueWithQuorum(quorum int) (T, bool) {
+	var zero T
+	for x, count := range c {
+		if count >= quorum {
+			return x, true
+		}
+	}
+	return zero, false
+}
+
 // figure out the most commonVersions across disk that satisfies
 // the 'writeQuorum' this function returns "" if quorum cannot
 // be achieved and disks have too many inconsistent versions.

cmd/erasure-metadata.go

@@ -344,9 +344,14 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.
 		return FileInfo{}, InsufficientReadQuorum{Err: errErasureReadQuorum, Type: RQInconsistentMeta}
 	}
 
-	// Find the successor mod time in quorum, otherwise leave the
+	// objProps represents properties that go beyond a single version
-	// candidate's successor modTime as found
+	type objProps struct {
-	succModTimeMap := make(map[time.Time]int)
+		succModTime time.Time
+		numVersions int
+	}
+	// Find the successor mod time and numVersions in quorum, otherwise leave the
+	// candidate as found
+	otherPropsMap := make(counterMap[objProps])
 	var candidate FileInfo
 	var found bool
 	for i, hash := range metaHashes {
@@ -356,24 +361,21 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.
 					candidate = metaArr[i]
 					found = true
 				}
-				succModTimeMap[metaArr[i].SuccessorModTime]++
+				props := objProps{
+					succModTime: metaArr[i].SuccessorModTime,
+					numVersions: metaArr[i].NumVersions,
+				}
+				otherPropsMap[props]++
 			}
 		}
 	}
-	var succModTime time.Time
-	var smodTimeQuorum bool
-	for smodTime, count := range succModTimeMap {
-		if count >= quorum {
-			smodTimeQuorum = true
-			succModTime = smodTime
-			break
-		}
-	}
 
 	if found {
-		if smodTimeQuorum {
+		// Update candidate FileInfo with succModTime and numVersions in quorum when available
-			candidate.SuccessorModTime = succModTime
+		if props, ok := otherPropsMap.GetValueWithQuorum(quorum); ok {
-			candidate.IsLatest = succModTime.IsZero()
+			candidate.SuccessorModTime = props.succModTime
+			candidate.IsLatest = props.succModTime.IsZero()
+			candidate.NumVersions = props.numVersions
 		}
 		return candidate, nil
 	}

cmd/erasure-metadata_test.go

@@ -158,7 +158,7 @@ func TestObjectToPartOffset(t *testing.T) {
 }
 
 func TestFindFileInfoInQuorum(t *testing.T) {
-	getNFInfo := func(n int, quorum int, t int64, dataDir string, succModTimes []time.Time) []FileInfo {
+	getNFInfo := func(n int, quorum int, t int64, dataDir string, succModTimes []time.Time, numVersions []int) []FileInfo {
 		fi := newFileInfo("test", 8, 8)
 		fi.AddObjectPart(1, "etag", 100, 100, UTCNow(), nil, nil)
 		fi.ModTime = time.Unix(t, 0)
@@ -171,6 +171,9 @@ func TestFindFileInfoInQuorum(t *testing.T) {
 				fis[i].SuccessorModTime = succModTimes[i]
 				fis[i].IsLatest = succModTimes[i].IsZero()
 			}
+			if numVersions != nil {
+				fis[i].NumVersions = numVersions[i]
+			}
 			quorum--
 			if quorum == 0 {
 				break
@@ -182,59 +185,86 @@ func TestFindFileInfoInQuorum(t *testing.T) {
 	commonSuccModTime := time.Date(2023, time.August, 25, 0, 0, 0, 0, time.UTC)
 	succModTimesInQuorum := make([]time.Time, 16)
 	succModTimesNoQuorum := make([]time.Time, 16)
+	commonNumVersions := 2
+	numVersionsInQuorum := make([]int, 16)
+	numVersionsNoQuorum := make([]int, 16)
 	for i := 0; i < 16; i++ {
 		if i < 4 {
 			continue
 		}
 		succModTimesInQuorum[i] = commonSuccModTime
+		numVersionsInQuorum[i] = commonNumVersions
 		if i < 9 {
 			continue
 		}
 		succModTimesNoQuorum[i] = commonSuccModTime
+		numVersionsNoQuorum[i] = commonNumVersions
 	}
 	tests := []struct {
 		fis                 []FileInfo
 		modTime             time.Time
 		succmodTimes        []time.Time
+		numVersions         []int
 		expectedErr         error
 		expectedQuorum      int
 		expectedSuccModTime time.Time
+		expectedNumVersions int
 		expectedIsLatest    bool
 	}{
 		{
-			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil),
+			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil, nil),
 			modTime:        time.Unix(1603863445, 0),
 			expectedErr:    nil,
 			expectedQuorum: 8,
 		},
 		{
-			fis:            getNFInfo(16, 7, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil),
+			fis:            getNFInfo(16, 7, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil, nil),
 			modTime:        time.Unix(1603863445, 0),
 			expectedErr:    InsufficientReadQuorum{},
 			expectedQuorum: 8,
 		},
 		{
-			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil),
+			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil, nil),
 			modTime:        time.Unix(1603863445, 0),
 			expectedErr:    InsufficientReadQuorum{},
 			expectedQuorum: 0,
 		},
 		{
-			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", succModTimesInQuorum),
+			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", succModTimesInQuorum, nil),
 			modTime:             time.Unix(1603863445, 0),
+			succmodTimes:        succModTimesInQuorum,
 			expectedErr:         nil,
 			expectedQuorum:      12,
 			expectedSuccModTime: commonSuccModTime,
 			expectedIsLatest:    false,
 		},
 		{
-			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", succModTimesNoQuorum),
+			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", succModTimesNoQuorum, nil),
 			modTime:             time.Unix(1603863445, 0),
+			succmodTimes:        succModTimesNoQuorum,
 			expectedErr:         nil,
 			expectedQuorum:      12,
 			expectedSuccModTime: time.Time{},
 			expectedIsLatest:    true,
 		},
+		{
+			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil, numVersionsInQuorum),
+			modTime:             time.Unix(1603863445, 0),
+			numVersions:         numVersionsInQuorum,
+			expectedErr:         nil,
+			expectedQuorum:      12,
+			expectedIsLatest:    true,
+			expectedNumVersions: 2,
+		},
+		{
+			fis:                 getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21", nil, numVersionsNoQuorum),
+			modTime:             time.Unix(1603863445, 0),
+			numVersions:         numVersionsNoQuorum,
+			expectedErr:         nil,
+			expectedQuorum:      12,
+			expectedIsLatest:    true,
+			expectedNumVersions: 0,
+		},
 	}
 
 	for _, test := range tests {
@@ -254,6 +284,11 @@ func TestFindFileInfoInQuorum(t *testing.T) {
 					t.Errorf("Expected IsLatest to be %v but got %v", test.expectedIsLatest, fi.IsLatest)
 				}
 			}
+			if test.numVersions != nil && test.expectedNumVersions > 0 {
+				if test.expectedNumVersions != fi.NumVersions {
+					t.Errorf("Expected Numversions to be %d but got %d", test.expectedNumVersions, fi.NumVersions)
+				}
+			}
 		})
 	}
 }