Merge pull request #332 from pgsty/claude/multi-pool-ilm-version-dup-210154

fix(ilm): count a version under old and healed headers as one revision

cmd/erasure-server-pool-lifecycle-authority_test.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"crypto/sha256"
+	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -29,6 +30,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cespare/xxhash/v2"
 	"github.com/google/uuid"
 	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/bucket/lifecycle"
@@ -632,6 +634,176 @@ func TestPoolsLifecycleLegacyHistory(t *testing.T) {
 	}
 }
 
+// authoritySignature2021 is the header signature of o, a version whose part
+// ETags are all empty, as the releases from RELEASE.2021-11-24 until
+// RELEASE.2022-01-25 computed it: with these ETags.
+func authoritySignature2021(o xlMetaV2Object) (sig [4]byte) {
+	o.ErasureIndex = 0
+	crc := hashDeterministicString(o.MetaUser) ^ hashDeterministicBytes(o.MetaSys)
+	o.MetaUser, o.MetaSys = nil, nil
+	bts, _ := o.MarshalMsg(nil)
+	crc ^= xxhash.Sum64(bts)
+	binary.LittleEndian.PutUint32(sig[:], uint32(crc^(crc>>32)))
+	return sig
+}
+
+// A drive holds the header that the release which wrote a version there
+// stored. Before RELEASE.2024-04-28 it has no erasure coding, and until
+// RELEASE.2022-01-25 its signature counted the empty part ETags of versions
+// written before RELEASE.2021-11-24. A heal writes the header of its own
+// release and leaves the other drives alone, so the drives of an old version
+// differ there for good once some were replaced. That is one revision, as
+// reads and heal take it, on however many of the drives.
+func TestPoolsLifecycleHeaderEras(t *testing.T) {
+	id := [16]byte{1}
+	current := func(drive int) xlMetaV2ShallowVersion { return authorityRevision(t, id, 1, drive+1, "") }
+	// stored returns the version as drive holds it after change, under the
+	// header that header derives from the one a current release writes.
+	stored := func(drive int, change func(*xlMetaV2Object), header func(xlMetaV2VersionHeader, xlMetaV2Object) xlMetaV2VersionHeader) xlMetaV2ShallowVersion {
+		ver := current(drive)
+		var v xlMetaV2Version
+		if _, err := v.UnmarshalMsg(ver.meta); err != nil {
+			t.Fatal(err)
+		}
+		change(v.ObjectV2)
+		meta, err := v.MarshalMsg(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return xlMetaV2ShallowVersion{header: header(ver.header, *v.ObjectV2), meta: meta}
+	}
+	eras := map[string]func(drive int) xlMetaV2ShallowVersion{
+		"no-erasure-coding": func(drive int) xlMetaV2ShallowVersion {
+			return stored(drive, func(*xlMetaV2Object) {}, func(h xlMetaV2VersionHeader, _ xlMetaV2Object) xlMetaV2VersionHeader {
+				h.EcN, h.EcM = 0, 0
+				return h
+			})
+		},
+		"empty-part-etags-signed": func(drive int) xlMetaV2ShallowVersion {
+			return stored(drive, func(o *xlMetaV2Object) { o.PartETags = []string{""} }, func(h xlMetaV2VersionHeader, o xlMetaV2Object) xlMetaV2VersionHeader {
+				h.EcN, h.EcM, h.Signature = 0, 0, authoritySignature2021(o)
+				return h
+			})
+		},
+	}
+	metaVs := slices.Repeat([]uint8{xlMetaVersion}, 16)
+	readQuorum := func(xlMetaV2ShallowVersion, uint8) int { return 12 }
+	for name, old := range eras {
+		t.Run(name, func(t *testing.T) {
+			if old(0).header == current(0).header {
+				t.Fatal("fixture: the old header is the current one")
+			}
+			if versionRevision(old(0), xlMetaVersion) != versionRevision(current(7), xlMetaVersion) {
+				t.Error("one version under the headers of two releases has two keys")
+			}
+			// The header of the key orders the merge: it is the current one.
+			if versionRevision(old(0), xlMetaVersion).header != current(0).header {
+				t.Error("the key does not carry the header of the current release")
+			}
+			for healed := range 17 {
+				metas := make([][]xlMetaV2ShallowVersion, 16)
+				for i := range metas {
+					if i < healed {
+						metas[i] = []xlMetaV2ShallowVersion{current(i)}
+					} else {
+						metas[i] = []xlMetaV2ShallowVersion{old(i)}
+					}
+				}
+				merged, err := authoritativeHistory(t.Context(), metas, metaVs, xlMetaVersion, 9, readQuorum)
+				if err != nil || len(merged) != 1 {
+					t.Errorf("%d healed drives: %d versions, err=%v", healed, len(merged), err)
+				}
+			}
+		})
+	}
+	// Control: the erasure coding still tells revisions apart, by the
+	// metadata. Another one under the same header is another revision.
+	other := stored(0, func(o *xlMetaV2Object) { o.ErasureM, o.ErasureN = 8, 8 }, func(h xlMetaV2VersionHeader, _ xlMetaV2Object) xlMetaV2VersionHeader { return h })
+	if versionRevision(other, xlMetaVersion) == versionRevision(current(0), xlMetaVersion) {
+		t.Error("another erasure coding under one header keeps the revision")
+	}
+}
+
+// The same on drives, through a heal: two versions with headers from before
+// RELEASE.2024-04-28 on a set where half of the drives were replaced. GET
+// reads them and heal has nothing left to do. Their expiry has to run as
+// well: of a noncurrent version with several pools, and of all versions of a
+// key with one pool too, which reads the history like the pools do. A
+// restore takes the drives for one generation, not for a failed status write.
+func TestPoolsLifecycleHealedOldHeaders(t *testing.T) {
+	for _, single := range []bool{false, true} {
+		name, key := "noncurrent-version", "ncv-old-headers"
+		if single {
+			name, key = "all-versions-one-pool", "eav-old-headers"
+		}
+		t.Run(name, func(t *testing.T) {
+			s := newSplitILM(t, versioning.Enabled, splitILMRules)
+			ctx := t.Context()
+			s.write(key, splitAt("old", 5, 0))
+			s.write(key, splitAt("new", 3, 0))
+			ids := []string{s.ids[key+" old"], s.ids[key+" new"]}
+			if single {
+				prior := s.z.serverPools
+				t.Cleanup(func() { s.z.serverPools = prior })
+				s.z.serverPools = s.z.serverPools[:1]
+			}
+			set := s.z.serverPools[0].getHashedSet(key)
+			disks := set.getDisks()
+			for _, disk := range disks {
+				rewriteTierTestMeta(t, disk, s.bucket, key, func(xl *xlMetaV2) {
+					for i := range xl.versions {
+						// Control: the key of a version this release wrote
+						// carries its header as stored.
+						if versionRevision(xl.versions[i], xl.metaV).header != xl.versions[i].header {
+							t.Fatal("the key of a current version carries another header")
+						}
+						xl.versions[i].header.EcN, xl.versions[i].header.EcM = 0, 0
+					}
+				})
+			}
+			replaced := len(disks) / 2
+			for _, disk := range disks[:replaced] {
+				if err := disk.Delete(ctx, s.bucket, key, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+					t.Fatal(err)
+				}
+				for _, id := range ids {
+					if _, err := s.z.HealObject(ctx, s.bucket, key, id, madmin.HealOpts{ScanMode: madmin.HealNormalScan}); err != nil {
+						t.Fatalf("heal %s: %v", id, err)
+					}
+				}
+			}
+			for i, disk := range disks {
+				if authorityHeader(t, disk, s.bucket, key, ids[0]).hasEC() != (i < replaced) {
+					t.Fatalf("fixture: drive %d of %d, %d replaced, holds the other header", i, len(disks), replaced)
+				}
+			}
+			authorityPresent(t, s, 0, key, ids[0])
+			res, err := s.z.HealObject(ctx, s.bucket, key, ids[0], madmin.HealOpts{DryRun: true, ScanMode: madmin.HealNormalScan})
+			if err != nil || slices.ContainsFunc(res.Before.Drives, func(d madmin.HealDriveInfo) bool { return d.State != madmin.DriveStateOk }) {
+				t.Fatalf("fixture: heal has work left: %v", err)
+			}
+
+			if gen, err := set.readRestoreGeneration(ctx, s.bucket, key, ids[0]); err != nil || gen.split {
+				t.Errorf("restore generation: split=%v, err=%v", gen.split, err)
+			}
+			if single {
+				_, err = s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll)
+			} else {
+				err = authorityExpireVersion(t, s, key, ids[0])
+			}
+			if err != nil {
+				t.Fatalf("expiry: %v", err)
+			}
+			if _, err := s.z.GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: ids[0]}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+				t.Errorf("the expired version: %v", err)
+			}
+			if !single {
+				authorityPresent(t, s, 0, key, ids[1])
+			}
+		})
+	}
+}
+
 // authorityDeleteAll is the expiry of all versions of a key as the expiry
 // worker sends it: a purge of all versions, approved by the lifecycle.
 var authorityDeleteAll = ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}}

cmd/erasure-server-pool-lifecycle.go

@@ -187,9 +187,17 @@ type revisionKey struct {
 // input is canonical: each map, including the part checksums of a legacy
 // version, is its own section with its entry count and sorted,
 // length-prefixed entries, since msgp encodes maps in iteration order and an
-// entry must not move between maps unnoticed. The header of a
+// entry must not move between maps unnoticed.
-// legacy version carries a signature over its per-shard checksums, so the
+//
-// key replaces it with one from the digest.
+// A drive holds the header that the release which wrote the version there
+// stored, and a heal writes the one of its own release: headers from before
+// RELEASE.2024-04-28 carry no erasure coding, and the signature counted
+// empty part ETags until RELEASE.2022-01-25. Reads and heal take such drives
+// for one version (matchesNotStrict), and nothing rewrites them. The key has
+// both as the current release computes them from the metadata, so that the
+// header of a version it wrote stays as stored: the signature also orders
+// versions of one time in the merge. The signature of a legacy version is
+// over its per-shard checksums; the key has one from the digest.
 func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey {
 	h := sha256.New()
 	writeUint := func(n int) {
@@ -267,8 +275,12 @@ func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey {
 			section("fields", 1)
 			writeField(bts)
 			h.Sum(key.digest[:0])
-			if v.Type == LegacyType {
+			switch v.Type {
+			case LegacyType:
 				copy(key.header.Signature[:], key.digest[:])
+			case ObjectType:
+				key.header.Signature = v.ObjectV2.Signature()
+				key.header.EcN, key.header.EcM = uint8(v.ObjectV2.ErasureN), uint8(v.ObjectV2.ErasureM)
 			}
 			return key
 		}
@@ -350,7 +362,7 @@ func authoritativeHistory(ctx context.Context, metas [][]xlMetaV2ShallowVersion,
 	// its count, and one that shares the header could stand in for the
 	// accepted one, so the merge only sees the accepted revision of each
 	// version and free versions as they are. It sees the headers of the
-	// keys, which do not differ between the drives of a legacy version.
+	// keys, which do not differ between the drives of a revision.
 	filtered := make([][]xlMetaV2ShallowVersion, len(metas))
 	for i, versions := range metas {
 		for j, ver := range versions {