From 0b1c21cb4b5250f43a106e815fb6fffe209a402f Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 25 Oct 2026 00:48:16 +0800 Subject: [PATCH] fix: retain unresolved generations during fast reads and healing Signed-off-by: Feng Ruohang --- cmd/erasure-healing.go | 3 + cmd/erasure-object.go | 41 +++++------ cmd/erasure-read-generation_test.go | 106 +++++++++++++++++++++++++++- cmd/metacache-null-quorum_test.go | 46 +++++++++--- 4 files changed, 161 insertions(+), 35 deletions(-) diff --git a/cmd/erasure-healing.go b/cmd/erasure-healing.go index 8fb371270..f6792aee7 100644 --- a/cmd/erasure-healing.go +++ b/cmd/erasure-healing.go @@ -373,6 +373,9 @@ func (er *erasureObjects) healObject(ctx context.Context, bucket string, object if err != nil { return result, err } + if err := checkObjectGenerationQuorum(latestMeta, partsMetadata, errs); err != nil { + return result, err + } // No modtime quorum filterDisksByETag := quorumETag != "" diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index c093bdfab..54b479e58 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -617,6 +617,18 @@ func readAllRawFileInfo(ctx context.Context, disks []StorageAPI, bucket, object return rawFileInfos, g.Wait() } +// An older quorum cannot resolve a later overwrite of the same version. +// Reads and healing must preserve that uncertainty rather than roll back +// the visible object or erase the newer fragments. +func checkObjectGenerationQuorum(fi FileInfo, metadata []FileInfo, errs []error) error { + for i, meta := range metadata { + if errs[i] == nil && meta.IsValid() && meta.VersionID == fi.VersionID && meta.ModTime.After(fi.ModTime) { + return InsufficientReadQuorum{Err: errErasureReadQuorum, Type: RQInconsistentMeta} + } + } + return nil +} + func pickLatestQuorumFilesInfo(ctx context.Context, rawFileInfos []RawFileInfo, errs []error, bucket, object string, readData, inclFreeVers bool) ([]FileInfo, []error) { metadataArray := make([]*xlMetaV2, len(rawFileInfos)) metaFileInfos := make([]FileInfo, len(rawFileInfos)) @@ -902,7 +914,6 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s } }() - validResp := 0 totalResp := 0 // minDisks value is only to reduce the number of calls @@ -935,15 +946,8 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s if err != nil { return FileInfo{}, nil, nil, time.Time{}, "", err } - for i, meta := range metaArr { - if errs[i] == nil && meta.IsValid() && meta.VersionID == fi.VersionID && meta.ModTime.After(fi.ModTime) { - // A quorum of an older generation does not resolve a later - // overwrite of the same version. Keep the uncertainty visible - // instead of silently rolling the object back on reads. - return FileInfo{}, nil, nil, time.Time{}, "", InsufficientReadQuorum{ - Err: errErasureReadQuorum, Type: RQInconsistentMeta, - } - } + if err := checkObjectGenerationQuorum(fi, metaArr, errs); err != nil { + return FileInfo{}, nil, nil, time.Time{}, "", err } onlineMeta := make([]FileInfo, len(metaArr)) @@ -965,11 +969,8 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s err error ) - for success := range done { + for range done { totalResp++ - if success { - validResp++ - } if totalResp >= minDisks && opts.FastGetObjInfo { rw.Lock() @@ -985,12 +986,9 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s } if totalResp < er.setDriveCount { - if !opts.FastGetObjInfo { - continue - } - if validResp < minDisks { - continue - } + // A fast quorum may contain only an older generation. Include + // slower metadata responses before selecting a readable value. + continue } rw.Lock() @@ -1003,9 +1001,6 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s fi, onlineMeta, onlineDisks, modTime, etag, err = calcQuorum(metaArr, errs) } rw.Unlock() - if err == nil && (fi.InlineData() || len(fi.Data) > 0) { - break - } } if err != nil { diff --git a/cmd/erasure-read-generation_test.go b/cmd/erasure-read-generation_test.go index 6ba917b6b..f734f2e90 100644 --- a/cmd/erasure-read-generation_test.go +++ b/cmd/erasure-read-generation_test.go @@ -22,8 +22,11 @@ import ( "context" "fmt" "net/http" + "sync" "testing" "time" + + madmin "github.com/minio/madmin-go/v3" ) // All metadata is valid and readable. Only the completed generations differ. @@ -124,10 +127,20 @@ func TestPoolsReadGenerationConflictDoesNotExposeOldValue(t *testing.T) { } } } + if tc.mode != "newer-quorum" { + for _, version := range []string{"", nullVersionID} { + for _, scan := range []madmin.HealScanMode{madmin.HealNormalScan, madmin.HealDeepScan} { + _, err := set.HealObject(ctx, bucket, object, version, madmin.HealOpts{ScanMode: scan, Remove: true}) + if !isErrReadQuorum(err) { + t.Errorf("heal version=%q scan=%v: want read-quorum error, got %v", version, scan, err) + } + } + } + } for i, disk := range disks { buf, err := disk.ReadAll(ctx, bucket, pathJoin(object, xlStorageFormatFile)) if err != nil || !bytes.Equal(buf, saved[i]) { - t.Errorf("read destroyed conflicting generation on disk %d: %v", i, err) + t.Errorf("read or heal destroyed conflicting generation on disk %d: %v", i, err) } } }) @@ -154,3 +167,94 @@ func TestPoolsReadUnavailablePoolDoesNotExposeOldValue(t *testing.T) { }) } } + +type generationReadGateDisk struct { + StorageAPI + ready chan<- struct{} + release <-chan struct{} +} + +func (d generationReadGateDisk) ReadXL(ctx context.Context, volume, path string, data bool) (RawFileInfo, error) { + fi, err := d.StorageAPI.ReadXL(ctx, volume, path, data) + d.ready <- struct{}{} + if d.release != nil { + <-d.release + } + return fi, err +} + +func (d generationReadGateDisk) ReadVersion(ctx context.Context, orig, volume, path, version string, opts ReadOptions) (FileInfo, error) { + fi, err := d.StorageAPI.ReadVersion(ctx, orig, volume, path, version, opts) + d.ready <- struct{}{} + if d.release != nil { + <-d.release + } + return fi, err +} + +func TestReadGenerationFastInfoWaitsForSlowDisk(t *testing.T) { + for _, version := range []string{"", nullVersionID} { + t.Run("version="+version, func(t *testing.T) { + z, bucket := consistencyPools(t) + const object = "slow-newer-generation" + set := z.serverPools[0].getHashedSet(object) + disks, switchDisks := installListingDiskSwitch(set) + defer switchDisks(disks) + base := UTCNow().Add(-time.Hour) + putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{MTime: base}) + saved := make([][]byte, len(disks)-1) + for i := range saved { + saved[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object) + } + putConsistencyObject(t, z, bucket, object, 0, "new", ObjectOptions{MTime: base.Add(time.Minute)}) + for i, meta := range saved { + if err := disks[i].WriteAll(t.Context(), bucket, pathJoin(object, xlStorageFormatFile), meta); err != nil { + t.Fatal(err) + } + } + ready, release := make(chan struct{}, len(disks)), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + wrapped := append([]StorageAPI(nil), disks...) + for i, disk := range disks { + gate := generationReadGateDisk{StorageAPI: disk, ready: ready} + if i == len(disks)-1 { + gate.release = release + } + wrapped[i] = gate + } + switchDisks(wrapped) + done := make(chan error, 1) + go func() { + _, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{FastGetObjInfo: true, VersionID: version}, false) + done <- err + }() + for range disks { + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("metadata reads did not reach the barrier") + } + } + early := false + select { + case err := <-done: + early = true + t.Errorf("fast read returned before the newer disk: %v", err) + case <-time.After(100 * time.Millisecond): + } + unblock() + if !early { + select { + case err := <-done: + if !isErrReadQuorum(err) { + t.Fatalf("fast read exposed the old generation: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("fast read did not finish after the newer disk") + } + } + }) + } +} diff --git a/cmd/metacache-null-quorum_test.go b/cmd/metacache-null-quorum_test.go index ee88f0cfa..d6f2d33ef 100644 --- a/cmd/metacache-null-quorum_test.go +++ b/cmd/metacache-null-quorum_test.go @@ -701,9 +701,13 @@ func TestSingleNullQuorumReadAndConsumers(t *testing.T) { } } put(newBody, oldTime.Add(time.Minute)) - newMinorityMeta := mustReadNullQuorumMeta(t, disks[14], bucket, object) + newMeta := make([][]byte, len(disks)) + for i, disk := range disks { + newMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object) + } // Model an incomplete overwrite using real inline shards: fifteen old // copies retain a read quorum; the final disk contains the newer minority. + // The old quorum does not prove that the newer overwrite was uncommitted. for i := range 15 { if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil { t.Fatal(err) @@ -721,18 +725,38 @@ func TestSingleNullQuorumReadAndConsumers(t *testing.T) { t.Fatalf("wrong readable generation: %v, %v", oi.ModTime, err) } } - checkRead(oldBody, oldTime) + for _, method := range []string{http.MethodGet, http.MethodHead} { + rec := nullQuorumRequest(t, router, method, getGetObjectURL("", bucket, object)) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("%s exposed an unresolved old quorum: %d", method, rec.Code) + } + } + if _, err := z.HealObject(t.Context(), bucket, object, "", madmin.HealOpts{Remove: true}); !isErrReadQuorum(err) { + t.Fatalf("heal accepted an unresolved old quorum: %v", err) + } // Reading must not reinterpret the minority as absent or delete it. minority, err := disks[15].ReadVersion(t.Context(), "", bucket, object, "", ReadOptions{}) if err != nil || !minority.ModTime.Equal(oldTime.Add(time.Minute)) { t.Fatalf("read mutated the minority: %+v, %v", minority, err) } + // Now model the safe direction: the latest generation has the quorum. + // Keep the migration, Walk and scanner assertions on that readable state. + for i := range 15 { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, newMeta[i]); err != nil { + t.Fatal(err) + } + } + if err := disks[15].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[15]); err != nil { + t.Fatal(err) + } + newTime := oldTime.Add(time.Minute) + checkRead(newBody, newTime) for _, kind := range []string{"rebalance", "decommission"} { t.Run(kind, func(t *testing.T) { var names []string consume := func(entry metaCacheEntry) { versions, err := entry.fileInfoVersions(bucket) - if err != nil || len(versions.Versions) != 1 || !versions.Versions[0].ModTime.Equal(oldTime) { + if err != nil || len(versions.Versions) != 1 || !versions.Versions[0].ModTime.Equal(newTime) { t.Errorf("invalid migration metadata: %+v, %v", versions, err) return } @@ -745,7 +769,7 @@ func TestSingleNullQuorumReadAndConsumers(t *testing.T) { } body, err := io.ReadAll(reader) reader.Close() - if err != nil || !bytes.Equal(body, oldBody) { + if err != nil || !bytes.Equal(body, newBody) { t.Errorf("migration could not read selected object data: %v", err) } names = append(names, entry.name) @@ -768,7 +792,7 @@ func TestSingleNullQuorumReadAndConsumers(t *testing.T) { } var names []string for result := range results { - if result.Err != nil || !result.Item.ModTime.Equal(oldTime) { + if result.Err != nil || !result.Item.ModTime.Equal(newTime) { t.Fatalf("Walk returned invalid metadata: %+v", result) } names = append(names, result.Item.Name) @@ -779,22 +803,22 @@ func TestSingleNullQuorumReadAndConsumers(t *testing.T) { } // Exercise the scanner's actual abandoned-child path. Make the scan drive - // miss the object, while keeping an older quorum and a newer minority on + // miss the object, while keeping the latest quorum and an older minority on // the remaining disks. Its queued heal must read all disks and repair both. - if err := disks[14].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, newMinorityMeta); err != nil { + if err := disks[14].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[14]); err != nil { t.Fatal(err) } if err := os.RemoveAll(filepath.Join(disks[15].Endpoint().Path, bucket, object)); err != nil { t.Fatal(err) } - scanNullQuorumAbandoned(t, z, bucket, object, oldTime) + scanNullQuorumAbandoned(t, z, bucket, object, newTime) for i, disk := range disks { fi, err := disk.ReadVersion(t.Context(), "", bucket, object, "", ReadOptions{Healing: true}) - if err != nil || !fi.ModTime.Equal(oldTime) { + if err != nil || !fi.ModTime.Equal(newTime) { t.Fatalf("scanner heal did not reconcile disk %d: %v, %v", i, fi.ModTime, err) } } - checkRead(oldBody, oldTime) + checkRead(newBody, newTime) put(newBody, oldTime.Add(2*time.Minute)) checkRead(newBody, oldTime.Add(2*time.Minute)) } @@ -890,5 +914,5 @@ func scanNullQuorumAbandoned(t *testing.T, z *erasureServerPools, bucket, object if flat == nil || flat.Objects != 1 || flat.Size != 8192 { t.Fatalf("scanner lost the healed object from usage: %+v", flat) } - t.Logf("scanner queued %v; healed old quorum, repaired minority/missing copies, and retained one 8192-byte object", healed) + t.Logf("scanner queued %v; healed latest quorum, repaired minority/missing copies, and retained one 8192-byte object", healed) }