Fix listing in objects split across pools (#19227)

Merging same-object - multiple versions from different pools would not always result in correct ordering.

When merging keep inputs separate.

```
λ mc ls --versions local/testbucket
------ before ------

[2024-03-05 20:17:19 CET]   228B STANDARD 1f163718-9bc5-4b01-bff7-5d8cf09caf10 v3 PUT hosts
[2024-03-05 20:19:56 CET]  19KiB STANDARD null v2 PUT hosts
[2024-03-05 20:17:15 CET]   228B STANDARD 73c9f651-f023-4566-b012-cc537fdb7ce2 v1 PUT hosts

------ after ------
λ mc ls --versions local/testbucket
[2024-03-05 20:19:56 CET]  19KiB STANDARD null v3 PUT hosts
[2024-03-05 20:17:19 CET]   228B STANDARD 1f163718-9bc5-4b01-bff7-5d8cf09caf10 v2 PUT hosts
[2024-03-05 20:17:15 CET]   228B STANDARD 73c9f651-f023-4566-b012-cc537fdb7ce2 v1 PUT hosts
```

cmd/metacache-entries.go

@@ -745,10 +745,10 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 
 		// Merge any unmerged
 		if len(toMerge) > 0 {
-			versions := make([]xlMetaV2ShallowVersion, 0, len(toMerge)+1)
+			versions := make([][]xlMetaV2ShallowVersion, 0, len(toMerge)+1)
 			xl, err := best.xlmeta()
 			if err == nil {
-				versions = append(versions, xl.versions...)
+				versions = append(versions, xl.versions)
 			}
 			for _, idx := range toMerge {
 				other := top[idx]
@@ -776,11 +776,12 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 						return err
 					}
 				}
-				versions = append(versions, xl2.versions...)
+				versions = append(versions, xl2.versions)
 			}
+
 			if xl != nil && len(versions) > 0 {
 				// Merge all versions. 'strict' doesn't matter since we only need one.
-				xl.versions = mergeXLV2Versions(readQuorum, true, 0, versions)
+				xl.versions = mergeXLV2Versions(readQuorum, true, 0, versions...)
 				if meta, err := xl.AppendTo(metaDataPoolGet()); err == nil {
 					if best.reusable {
 						metaDataPoolPut(best.metadata)

cmd/xl-storage-format-v2.go

@@ -1879,6 +1879,7 @@ func (x xlMetaV2) ListVersions(volume, path string, allParts bool) ([]FileInfo,
 
 // mergeXLV2Versions will merge all versions, typically from different disks
 // that have at least quorum entries in all metas.
+// Each version slice should be sorted.
 // Quorum must be the minimum number of matching metadata files.
 // Quorum should be > 1 and <= len(versions).
 // If strict is set to false, entries that match type

cmd/xl-storage-format-v2_test.go

@@ -21,6 +21,7 @@ import (
 	"bufio"
 	"bytes"
 	"compress/gzip"
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -1011,6 +1012,77 @@ func Test_mergeXLV2Versions2(t *testing.T) {
 	}
 }
 
+func Test_mergeEntryChannels(t *testing.T) {
+	dataZ, err := os.ReadFile("testdata/xl-meta-merge.zip")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var vers []metaCacheEntry
+	zr, err := zip.NewReader(bytes.NewReader(dataZ), int64(len(dataZ)))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, file := range zr.File {
+		if file.UncompressedSize64 == 0 {
+			continue
+		}
+		in, err := file.Open()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer in.Close()
+		buf, err := io.ReadAll(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		buf = xlMetaV2TrimData(buf)
+
+		vers = append(vers, metaCacheEntry{
+			name:     "a",
+			metadata: buf,
+		})
+	}
+
+	// Shuffle...
+	for i := 0; i < 100; i++ {
+		rng := rand.New(rand.NewSource(int64(i)))
+		rng.Shuffle(len(vers), func(i, j int) {
+			vers[i], vers[j] = vers[j], vers[i]
+		})
+		var entries []chan metaCacheEntry
+		for _, v := range vers {
+			v.cached = nil
+			ch := make(chan metaCacheEntry, 1)
+			ch <- v
+			close(ch)
+			entries = append(entries, ch)
+		}
+		out := make(chan metaCacheEntry, 1)
+		err := mergeEntryChannels(context.Background(), entries, out, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		got, ok := <-out
+		if !ok {
+			t.Fatal("Got no result")
+		}
+
+		xl, err := got.xlmeta()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(xl.versions) != 3 {
+			t.Fatal("Got wrong number of versions, want 3, got", len(xl.versions))
+		}
+		if !sort.SliceIsSorted(xl.versions, func(i, j int) bool {
+			return xl.versions[i].header.sortsBefore(xl.versions[j].header)
+		}) {
+			t.Errorf("Got unsorted result")
+		}
+	}
+}
+
 func TestXMinIOHealingSkip(t *testing.T) {
 	xl := xlMetaV2{}
 	failOnErr := func(err error) {