mirror of
https://github.com/pgsty/minio.git
synced 2026-11-12 10:35:33 +03:00
fix(restore): ignore derived data-dir flag in revision identity
Keep restored versions consistent across expiry and successive heals while preserving metadata, layout and non-object revision distinctions. Fixes #344 Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -696,8 +696,10 @@ func TestPoolsLifecycleHeaderEras(t *testing.T) {
|
||||
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 {
|
||||
// The normalized current header orders the merge.
|
||||
wantHeader := current(0).header
|
||||
wantHeader.Flags &^= xlFlagUsesDataDir
|
||||
if versionRevision(old(0), xlMetaVersion).header != wantHeader {
|
||||
t.Error("the key does not carry the header of the current release")
|
||||
}
|
||||
for healed := range 17 {
|
||||
@@ -753,8 +755,10 @@ func TestPoolsLifecycleHealedOldHeaders(t *testing.T) {
|
||||
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 {
|
||||
// carries its header with only the time-derived flag removed.
|
||||
wantHeader := xl.versions[i].header
|
||||
wantHeader.Flags &^= xlFlagUsesDataDir
|
||||
if versionRevision(xl.versions[i], xl.metaV).header != wantHeader {
|
||||
t.Fatal("the key of a current version carries another header")
|
||||
}
|
||||
xl.versions[i].header.EcN, xl.versions[i].header.EcM = 0, 0
|
||||
|
||||
@@ -195,8 +195,9 @@ type revisionKey struct {
|
||||
// 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
|
||||
// header of a version it wrote stays as stored, except UsesDataDir, which can
|
||||
// change solely because a restored copy expired before healing. 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()
|
||||
@@ -281,6 +282,7 @@ func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey {
|
||||
case ObjectType:
|
||||
key.header.Signature = v.ObjectV2.Signature()
|
||||
key.header.EcN, key.header.EcM = uint8(v.ObjectV2.ErasureN), uint8(v.ObjectV2.ErasureM)
|
||||
key.header.Flags &^= xlFlagUsesDataDir
|
||||
}
|
||||
return key
|
||||
}
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
// Copyright (c) 2026 Feng Ruohang
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/minio/madmin-go/v3"
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
"github.com/minio/minio/internal/bucket/versioning"
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
)
|
||||
|
||||
func TestVersionRevisionRestoreFlags(t *testing.T) {
|
||||
base := authorityRevision(t, [16]byte{1}, 1, 1, "")
|
||||
cleared := base
|
||||
cleared.header.Flags &^= xlFlagUsesDataDir
|
||||
set := base
|
||||
set.header.Flags |= xlFlagUsesDataDir
|
||||
if versionRevision(cleared, xlMetaVersion) != versionRevision(set, xlMetaVersion) {
|
||||
t.Fatal("derived flag splits one revision")
|
||||
}
|
||||
for _, change := range []struct {
|
||||
name string
|
||||
apply func(*xlMetaV2Object)
|
||||
}{
|
||||
{"data-dir", func(o *xlMetaV2Object) { o.DataDir[0]++ }},
|
||||
{"restore", func(o *xlMetaV2Object) { o.MetaUser[xhttp.AmzRestore] = `ongoing-request="true"` }},
|
||||
{"retention", func(o *xlMetaV2Object) { o.MetaUser["x-amz-object-lock-mode"] = "COMPLIANCE" }},
|
||||
{"layout", func(o *xlMetaV2Object) { o.ErasureM, o.ErasureN = 8, 8 }},
|
||||
} {
|
||||
t.Run(change.name, func(t *testing.T) {
|
||||
var v xlMetaV2Version
|
||||
if _, err := v.unmarshalV(xlMetaVersion, base.meta); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
change.apply(v.ObjectV2)
|
||||
meta, err := v.MarshalMsg(nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
different := xlMetaV2ShallowVersion{header: cleared.header, meta: meta}
|
||||
if versionRevision(different, xlMetaVersion) == versionRevision(set, xlMetaVersion) {
|
||||
t.Fatal("real metadata change lost")
|
||||
}
|
||||
})
|
||||
}
|
||||
// Every other header flag remains part of identity, even if metadata is equal.
|
||||
for flag := xlFlags(1); flag != 0; flag <<= 1 {
|
||||
if flag == xlFlagUsesDataDir {
|
||||
continue
|
||||
}
|
||||
other := base
|
||||
other.header.Flags ^= flag
|
||||
if versionRevision(other, xlMetaVersion) == versionRevision(base, xlMetaVersion) {
|
||||
t.Fatalf("flag %d was normalized", flag)
|
||||
}
|
||||
}
|
||||
malformed := base
|
||||
malformed.meta = []byte{0xc1}
|
||||
other := malformed
|
||||
other.header.Flags ^= xlFlagUsesDataDir
|
||||
if versionRevision(malformed, xlMetaVersion) == versionRevision(other, xlMetaVersion) {
|
||||
t.Fatal("undecodable metadata was normalized")
|
||||
}
|
||||
marker := authorityMarker(t, [16]byte{2}, 2)
|
||||
other = marker
|
||||
other.header.Flags ^= xlFlagUsesDataDir
|
||||
if versionRevision(marker, xlMetaVersion) == versionRevision(other, xlMetaVersion) {
|
||||
t.Fatal("delete marker was normalized as an object")
|
||||
}
|
||||
}
|
||||
|
||||
// Two drives are replaced one at a time after the restore expires. The old
|
||||
// headers describe the pre-expiry state; heal writes the current derived flag.
|
||||
func TestRestoreExpiryAcrossSuccessiveHeals(t *testing.T) {
|
||||
r := newRestoreSafetyTest(t, 2, versioning.Enabled)
|
||||
ctx := t.Context()
|
||||
oi := r.put("restored", false, true)
|
||||
r.restore(oi)
|
||||
before := r.fileInfo(oi.VersionID)
|
||||
disks := r.z.serverPools[0].getHashedSet(r.object).getDisks()
|
||||
// Model the headers written before wall-clock expiry without a timing race.
|
||||
for _, disk := range disks {
|
||||
rewriteTierTestMeta(t, disk, r.bucket, r.object, func(xl *xlMetaV2) { xl.versions[0].header.Flags |= xlFlagUsesDataDir })
|
||||
}
|
||||
for _, i := range []int{3, 2} {
|
||||
if err := disks[i].Delete(ctx, r.bucket, r.object, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := r.z.HealObject(ctx, r.bucket, r.object, oi.VersionID, madmin.HealOpts{ScanMode: madmin.HealNormalScan}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for i, disk := range disks {
|
||||
raw, err := disk.ReadXL(ctx, r.bucket, r.object, false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var xl xlMetaV2
|
||||
if err = xl.Load(raw.Buf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := xl.versions[0].header.Flags&xlFlagUsesDataDir != 0; got != (i < 2) {
|
||||
t.Fatalf("drive %d has unexpected derived flag %v", i, got)
|
||||
}
|
||||
}
|
||||
if _, _, err := r.z.lifecycleEventsAllPools(ctx, r.bucket, r.object); err != nil {
|
||||
t.Fatalf("history with split flags: %v", err)
|
||||
}
|
||||
current, err := r.z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(oi.VersionID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = expireTransitionedObject(ctx, r.z, ¤t, lifecycle.Event{Action: lifecycle.DeleteRestoredAction}, lcEventSrc_Scanner); err != nil {
|
||||
t.Fatalf("expire restored copy: %v", err)
|
||||
}
|
||||
if n := r.dataDirs(before.DataDir); n != 0 {
|
||||
t.Fatalf("restore cleanup left %d data directories", n)
|
||||
}
|
||||
if got := readCopyChecksumObject(t, r.z, r.bucket, r.object, r.opts(oi.VersionID)); string(got) != string(r.bodies["restored"]) {
|
||||
t.Fatal("tiered object is no longer readable")
|
||||
}
|
||||
r.put("newer", false, false)
|
||||
if _, _, err = r.z.lifecycleEventsAllPools(ctx, r.bucket, r.object); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = r.z.deleteAllHistory(ctx, r.bucket, r.object); err != nil {
|
||||
t.Fatalf("purge history with split flags: %v", err)
|
||||
}
|
||||
// The rule permits tiering, not a purge. Normalizing the header must not
|
||||
// turn a readable history into permission to remove the versions.
|
||||
if _, err = r.z.DeleteObject(ctx, r.bucket, r.object, ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}}); !errors.Is(err, errLifecycleNotApplicable) {
|
||||
t.Fatalf("purge without an applicable rule: %v", err)
|
||||
}
|
||||
if _, err = r.z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(oi.VersionID)); err != nil {
|
||||
t.Fatalf("refused purge removed the tiered version: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -320,11 +320,8 @@ func (er erasureObjects) stageRestore(ctx context.Context, bucket, object string
|
||||
|
||||
// fileInfoRevisions are the revisions the entry that writing fi creates can
|
||||
// have, as readRestoreGeneration and the convergence identify the entries
|
||||
// they read. The header flag that tells whether a transitioned version uses
|
||||
// its data directory follows whether its restored copy has expired at the
|
||||
// moment a drive writes the entry, so a restore whose copy expires while it
|
||||
// publishes can find either value: both are this entry. Every other field of
|
||||
// the header and the full metadata digest still have to match.
|
||||
// they read. versionRevision normalizes the time-derived UsesDataDir flag;
|
||||
// the stored data directory and restore metadata still have to match.
|
||||
func fileInfoRevisions(fi FileInfo) ([]revisionKey, error) {
|
||||
var x xlMetaV2
|
||||
if err := x.AddVersion(fi); err != nil {
|
||||
@@ -342,9 +339,7 @@ func fileInfoRevisions(fi FileInfo) ([]revisionKey, error) {
|
||||
return nil, errFileCorrupt
|
||||
}
|
||||
ver := y.versions[0]
|
||||
other := ver
|
||||
other.header.Flags ^= xlFlagUsesDataDir
|
||||
return []revisionKey{versionRevision(ver, y.metaV), versionRevision(other, y.metaV)}, nil
|
||||
return []revisionKey{versionRevision(ver, y.metaV)}, nil
|
||||
}
|
||||
|
||||
// restorePublication remembers, for a commit that published the staged copy
|
||||
|
||||
@@ -731,7 +731,7 @@ func TestRestorePublishedRevisionAcrossExpiry(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stored := func() revisionKey {
|
||||
stored := func() (xlMetaV2VersionHeader, revisionKey) {
|
||||
var x xlMetaV2
|
||||
if err := x.AddVersion(published); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -744,16 +744,19 @@ func TestRestorePublishedRevisionAcrossExpiry(t *testing.T) {
|
||||
if err := y.Load(buf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return versionRevision(y.versions[0], y.metaV)
|
||||
return y.versions[0].header, versionRevision(y.versions[0], y.metaV)
|
||||
}
|
||||
before := stored()
|
||||
beforeHeader, before := stored()
|
||||
time.Sleep(time.Until(expiry) + 100*time.Millisecond)
|
||||
after := stored()
|
||||
if before == after {
|
||||
afterHeader, after := stored()
|
||||
if beforeHeader.Flags == afterHeader.Flags {
|
||||
t.Fatal("fixture: the stored flag did not change at the expiry")
|
||||
}
|
||||
if before != after {
|
||||
t.Fatal("expiry changed the identity of the published revision")
|
||||
}
|
||||
for _, got := range []revisionKey{before, after} {
|
||||
if got != keys[0] && got != keys[1] {
|
||||
if got != keys[0] {
|
||||
t.Errorf("a publication's own entry is not recognized: %v not in %v", got, keys)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user