From ea7c794a941bec360b6d1bdb3e0ee55e824d01fa Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 25 Oct 2026 12:22:47 +0800 Subject: [PATCH] fix: preserve version state across late rollback and nested mutations Signed-off-by: Feng Ruohang --- cmd/erasure-healing-common.go | 2 +- cmd/erasure-metadata.go | 97 +++----- cmd/erasure-object.go | 10 +- cmd/object-api-utils.go | 2 +- cmd/object-mutation-consistency_test.go | 212 +++++++++++++++-- cmd/object-mutation-lease_test.go | 287 ++++++++++++++++++++++++ cmd/object-mutation.go | 204 +++++++++++++++++ cmd/rename-cancellation_test.go | 11 +- cmd/xl-storage-format-v2.go | 4 + cmd/xl-storage.go | 53 ++--- 10 files changed, 777 insertions(+), 105 deletions(-) create mode 100644 cmd/object-mutation-lease_test.go create mode 100644 cmd/object-mutation.go diff --git a/cmd/erasure-healing-common.go b/cmd/erasure-healing-common.go index 0bde3107d..fd204383d 100644 --- a/cmd/erasure-healing-common.go +++ b/cmd/erasure-healing-common.go @@ -185,7 +185,7 @@ func listObjectModtimes(partsMetadata []FileInfo, errs []error) (modTimes []time func filterOnlineDisksInplace(fi FileInfo, partsMetadata []FileInfo, onlineDisks []StorageAPI) { for i, meta := range partsMetadata { - if fi.XLV1 == meta.XLV1 { + if sameFileInfoVersion(fi, meta) { continue } onlineDisks[i] = nil diff --git a/cmd/erasure-metadata.go b/cmd/erasure-metadata.go index 1c3209f24..5f1c92ac2 100644 --- a/cmd/erasure-metadata.go +++ b/cmd/erasure-metadata.go @@ -19,7 +19,6 @@ package cmd import ( "context" - "encoding/hex" "fmt" "sort" "strings" @@ -28,8 +27,6 @@ import ( "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/bucket/lifecycle" "github.com/minio/minio/internal/bucket/replication" - "github.com/minio/minio/internal/crypto" - "github.com/minio/minio/internal/hash/sha256" xhttp "github.com/minio/minio/internal/http" "github.com/pgsty/silo-pkg/v3/sync/errgroup" ) @@ -319,68 +316,48 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time. if quorum < 1 { return FileInfo{}, InsufficientReadQuorum{Err: errErasureReadQuorum, Type: RQInsufficientOnlineDrives} } - metaHashes := make([]string, len(metaArr)) - h := sha256.New() + // Group by exact logical state rather than a layout-only hash. There are + // at most one erasure set's worth of candidates; healthy objects compare + // against a single representative without hashing or a second XL decode. + groups := make([]int, len(metaArr)) + representatives := make([]int, 0, len(metaArr)) + counts := make([]int, len(metaArr)+1) for i, meta := range metaArr { if !meta.IsValid() { continue } - etagOnly := modTime.Equal(timeSentinel) && (etag != "" && etag == meta.Metadata["etag"]) - mtimeValid := meta.ModTime.Equal(modTime) - if mtimeValid || etagOnly { - fmt.Fprintf(h, "%v", meta.XLV1) - for _, part := range meta.Parts { - fmt.Fprintf(h, "part.%d", part.Number) - fmt.Fprintf(h, "part.%d", part.Size) - } - // Previously we checked if we had quorum on DataDir value. - // We have removed this check to allow reading objects with different DataDir - // values in a few drives (due to a rebalance-stop race bug) - // provided their their etags or ModTimes match. - - if !meta.Deleted && meta.Size != 0 { - fmt.Fprintf(h, "%v+%v", meta.Erasure.DataBlocks, meta.Erasure.ParityBlocks) - fmt.Fprintf(h, "%v", meta.Erasure.Distribution) - } - - if meta.IsRemote() { - // ILM transition fields - fmt.Fprint(h, meta.TransitionStatus) - fmt.Fprint(h, meta.TransitionTier) - fmt.Fprint(h, meta.TransitionedObjName) - fmt.Fprint(h, meta.TransitionVersionID) - } - - // If metadata says encrypted, ask for it in quorum. - if etyp, ok := crypto.IsEncrypted(meta.Metadata); ok { - fmt.Fprint(h, etyp) - } - - // If compressed, look for compressed FileInfo only - if meta.IsCompressed() { - fmt.Fprint(h, meta.Metadata[ReservedMetadataPrefix+"compression"]) - } - - metaHashes[i] = hex.EncodeToString(h.Sum(nil)) - h.Reset() - } - } - - metaHashCountMap := make(map[string]int) - for _, hash := range metaHashes { - if hash == "" { + etagOnly := modTime.Equal(timeSentinel) && etag != "" && etag == meta.Metadata["etag"] + if !meta.ModTime.Equal(modTime) && !etagOnly { continue } - metaHashCountMap[hash]++ - } - - maxHash := "" - maxCount := 0 - for hash, count := range metaHashCountMap { - if count > maxCount { - maxCount = count - maxHash = hash + group := 0 + for j, representative := range representatives { + if sameFileInfoVersion(metaArr[representative], meta) { + group = j + 1 + break + } } + if group == 0 { + representatives = append(representatives, i) + group = len(representatives) + } + groups[i] = group + counts[group]++ + } + maxGroup, maxCount := 0, 0 + tied := false + for group, count := range counts { + if group == 0 { + continue + } + if count > maxCount { + maxGroup, maxCount, tied = group, count, false + } else if count == maxCount { + tied = true + } + } + if tied { + return FileInfo{}, InsufficientReadQuorum{Err: errErasureReadQuorum, Type: RQInconsistentMeta} } if maxCount < quorum { @@ -397,8 +374,8 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time. otherPropsMap := make(counterMap[objProps]) var candidate FileInfo var found bool - for i, hash := range metaHashes { - if hash == maxHash { + for i, group := range groups { + if group == maxGroup { if metaArr[i].IsValid() { if !found { candidate = metaArr[i] diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 035fa456c..e9a2fbb8f 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -997,7 +997,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s rw.Lock() // when its a versioned bucket and empty versionID - at totalResp == setDriveCount // we must use rawFileInfo to resolve versions to figure out the latest version. - if opts.VersionID == "" && totalResp == er.setDriveCount { + if opts.VersionID == "" && totalResp == er.setDriveCount && !uniformLatestVersion(rawArr, metaArr, errs) { fi, onlineMeta, onlineDisks, modTime, etag, err = calcQuorum(pickLatestQuorumFilesInfo(ctx, rawArr, errs, bucket, object, readData, opts.InclFreeVersions)) } else { @@ -1103,11 +1103,19 @@ func (er erasureObjects) getObjectInfoAndQuorum(ctx context.Context, bucket, obj // Similar to rename but renames data from srcEntry to dstEntry at dataDir func renameData(ctx context.Context, disks []StorageAPI, srcBucket, srcEntry string, metadata []FileInfo, dstBucket, dstEntry string, writeQuorum int) ([]StorageAPI, []byte, string, error) { + if err := ctx.Err(); err != nil { + return nil, nil, "", err + } + // Admission is cancelable. After admission use actual disk results for + // commit/rollback, even if the client or namespace lease is canceled. + ctx = context.WithoutCancel(ctx) g := errgroup.WithNErrs(len(disks)) fvID := mustGetUUID() for index := range disks { + metadata[index].Metadata = maps.Clone(metadata[index].Metadata) metadata[index].SetTierFreeVersionID(fvID) + metadata[index].Metadata[rollbackIDKey] = fvID } diskVersions := make([][]byte, len(disks)) diff --git a/cmd/object-api-utils.go b/cmd/object-api-utils.go index e8648b725..be9fd14f1 100644 --- a/cmd/object-api-utils.go +++ b/cmd/object-api-utils.go @@ -404,7 +404,7 @@ func cleanMetadata(metadata map[string]string) map[string]string { // Remove STANDARD StorageClass metadata = removeStandardStorageClass(metadata) // Clean meta etag keys 'md5Sum', 'etag', "expires", "x-amz-tagging". - return cleanMetadataKeys(metadata, "md5Sum", "etag", "expires", xhttp.AmzObjectTagging, "last-modified", VersionPurgeStatusKey) + return cleanMetadataKeys(metadata, "md5Sum", "etag", "expires", xhttp.AmzObjectTagging, "last-modified", VersionPurgeStatusKey, rollbackIDKey) } // Filter X-Amz-Storage-Class field only if it is set to STANDARD. diff --git a/cmd/object-mutation-consistency_test.go b/cmd/object-mutation-consistency_test.go index b042c9917..d4b889641 100644 --- a/cmd/object-mutation-consistency_test.go +++ b/cmd/object-mutation-consistency_test.go @@ -20,10 +20,6 @@ package cmd import ( "bytes" "context" - "github.com/minio/minio/internal/cachevalue" - "github.com/minio/minio/internal/config/drive" - xhttp "github.com/minio/minio/internal/http" - xioutil "github.com/minio/minio/internal/ioutil" "io" "os" "strconv" @@ -31,6 +27,11 @@ import ( "sync" "testing" "time" + + "github.com/minio/minio/internal/cachevalue" + "github.com/minio/minio/internal/config/drive" + xhttp "github.com/minio/minio/internal/http" + xioutil "github.com/minio/minio/internal/ioutil" ) func mutationPools(t testing.TB) (*erasureServerPools, string) { @@ -58,6 +59,7 @@ func mutationPools(t testing.TB) (*erasureServerPools, string) { } return z, bucket } + func mutationPut(t testing.TB, z *erasureServerPools, bucket, obj string, pool int, body string, when time.Time) { t.Helper() _, err := z.serverPools[pool].PutObject(context.Background(), bucket, obj, mustGetPutObjReader(t, bytes.NewBufferString(body), int64(len(body)), "", ""), ObjectOptions{MTime: when}) @@ -92,13 +94,16 @@ func TestMutationReadPreservesAcknowledgedTags(t *testing.T) { t.Fatal(err) } t.Logf("old/new tags=%q/%q; equal tuple=%t", before.Metadata[xhttp.AmzObjectTagging], after.Metadata[xhttp.AmzObjectTagging], before.ModTime.Equal(after.ModTime) && before.DataDir == after.DataDir && before.Metadata["etag"] == after.Metadata["etag"]) - oi, err := set.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{FastGetObjInfo: true}) - if err != nil { - t.Fatal(err) - } - t.Logf("successful fast GET metadata returned tags=%q", oi.UserTags) - if oi.UserTags != tags { - t.Fatalf("acknowledged tags were replaced by stale metadata despite matching ModTime/DataDir/ETag: %q", oi.UserTags) + for _, fast := range []bool{false, true} { + for _, version := range []string{"", nullVersionID} { + oi, err := set.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{FastGetObjInfo: fast, VersionID: version}) + if err != nil { + t.Fatal(err) + } + if oi.UserTags != tags { + t.Fatalf("acknowledged tags replaced by stale metadata (fast=%t version=%q): %q", fast, version, oi.UserTags) + } + } } } @@ -196,14 +201,14 @@ func TestMutationRollbackPreservesAcknowledgedTags(t *testing.T) { } } -type reviewPartialRename struct { +type delayedRenameResult struct { StorageAPI fail bool entered chan<- struct{} release <-chan struct{} } -func (d reviewPartialRename) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, opts RenameOptions) (RenameDataResp, error) { +func (d delayedRenameResult) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, opts RenameOptions) (RenameDataResp, error) { if d.fail { return RenameDataResp{}, errDiskNotFound } @@ -221,7 +226,7 @@ func (d reviewPartialRename) RenameData(ctx context.Context, sv, sp string, fi F func TestMutationRollbackPreservesConcurrentVersion(t *testing.T) { z, bucket := mutationPools(t) const object = "rollback-and-version" - size, _ := strconv.Atoi(os.Getenv("SILO_REVIEW_BYTES")) + size, _ := strconv.Atoi(os.Getenv("SILO_TEST_MUTATION_BYTES")) payload := func(v string) string { if size > len(v) { return v + strings.Repeat(".", size-len(v)) @@ -239,7 +244,7 @@ func TestMutationRollbackPreservesConcurrentVersion(t *testing.T) { defer unblock() aDisks := append([]StorageAPI(nil), disks...) for i, d := range disks { - aDisks[i] = reviewPartialRename{d, i >= 4, entered, release} + aDisks[i] = delayedRenameResult{d, i >= 4, entered, release} } swap(aDisks) aDone := make(chan error, 1) @@ -465,3 +470,180 @@ func TestMutationDeleteJoinsNestedWorker(t *testing.T) { } } } + +// Even an update that leaves all public fields unchanged is a later successful +// operation. An old writer must not regain undo ownership through an ABA cycle. +func TestMutationNoopUpdateRevokesUndo(t *testing.T) { + for _, method := range []string{"update", "write", "aba"} { + t.Run(method, func(t *testing.T) { + z, bucket := mutationPools(t) + const object = "no-op-update" + mutationPut(t, z, bucket, object, 0, "acknowledged", time.Now()) + disks := z.serverPools[0].getHashedSet(object).getDisks() + for _, disk := range disks { + fi, err := disk.ReadVersion(t.Context(), "", bucket, object, "", ReadOptions{ReadData: true}) + if err != nil { + t.Fatal(err) + } + switch method { + case "update": + err = disk.UpdateMetadata(t.Context(), bucket, object, fi, UpdateMetadataOpts{}) + case "write": + err = disk.WriteMetadata(t.Context(), "", bucket, object, fi) + case "aba": + for _, etag := range []string{"temporary", fi.Metadata["etag"]} { + err = disk.UpdateMetadata(t.Context(), bucket, object, FileInfo{Metadata: map[string]string{"etag": etag}}, UpdateMetadataOpts{}) + if err != nil { + t.Fatal(err) + } + } + } + if err != nil { + t.Fatal(err) + } + if err := disk.DeleteVersion(t.Context(), bucket, object, fi, false, DeleteOptions{UndoWrite: true}); err != errFileVersionNotFound { + t.Fatalf("old writer retained undo after %s: %v", method, err) + } + } + r, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + body, err := io.ReadAll(r) + if err != nil || string(body) != "acknowledged" { + t.Fatalf("later acknowledged state lost: %q, %v", body, err) + } + }) + } +} + +func TestMutationCanceledSuccessfulCommitIsNotUndone(t *testing.T) { + z, bucket := mutationPools(t) + const object = "cancel-after-commit" + mutationPut(t, z, bucket, object, 0, "initial", time.Now().Add(-time.Hour)) + set := z.serverPools[0].getHashedSet(object) + disks, swap := installListingDiskSwitch(set) + defer swap(disks) + entered, 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 { + wrapped[i] = delayedRenameResult{disk, false, entered, release} + } + swap(wrapped) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, strings.NewReader("committed"), 9, "", ""), ObjectOptions{}) + done <- err + }() + for range disks { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("physical commit barrier not reached") + } + } + cancel() + unblock() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("commit did not finish") + } + swap(disks) + r, err := set.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + body, err := io.ReadAll(r) + if err != nil || string(body) != "committed" { + t.Fatalf("cancellation rolled back an actual successful quorum: %q, %v", body, err) + } +} + +type orderedRenameDisk struct { + StorageAPI + early bool + earlyDone chan<- struct{} + release <-chan struct{} +} + +func (d orderedRenameDisk) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, opts RenameOptions) (RenameDataResp, error) { + if !d.early { + <-d.release + } + r, err := d.StorageAPI.RenameData(ctx, sv, sp, fi, dv, dp, opts) + if d.early { + d.earlyDone <- struct{}{} + <-d.release + } + return r, err +} + +func TestMutationCrossDiskOrderHasStableRead(t *testing.T) { + z, bucket := mutationPools(t) + const object = "cross-disk-order" + base := time.Now().Add(-time.Hour) + mutationPut(t, z, bucket, object, 0, "initial", base) + set := z.serverPools[0].getHashedSet(object) + disks, swap := installListingDiskSwitch(set) + defer swap(disks) + earlyDone, release := make(chan struct{}, 4), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + aDisks := append([]StorageAPI(nil), disks...) + for i, d := range disks { + aDisks[i] = orderedRenameDisk{d, i < 4, earlyDone, release} + } + swap(aDisks) + put := func(body string, mtime time.Time) error { + _, e := z.serverPools[0].PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, strings.NewReader(body), int64(len(body)), "", ""), + ObjectOptions{NoLock: true, MTime: mtime}) + return e + } + aDone := make(chan error, 1) + go func() { aDone <- put("writer-A", base.Add(time.Minute)) }() + for range 4 { + select { + case <-earlyDone: + case <-time.After(5 * time.Second): + t.Fatal("A did not reach first-half barrier") + } + } + swap(disks) + if err := put("writer-B", base.Add(2*time.Minute)); err != nil { + t.Fatalf("B failed: %v", err) + } + unblock() + select { + case err := <-aDone: + if err != nil { + t.Fatalf("A failed: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("A did not finish") + } + t.Log("both PUTs succeeded; final per-disk order is A→B on four disks and B→A on four disks; no writer remains") + swap(disks) + for range 8 { + for _, version := range []string{"", nullVersionID} { + r, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{FastGetObjInfo: true, VersionID: version}) + if err != nil { + t.Fatalf("quiescent successful writes became unreadable: %v", err) + } + body, err := io.ReadAll(r) + r.Close() + if err != nil || string(body) != "writer-B" { + t.Fatalf("completed reads changed value: %q %v", body, err) + } + } + } +} diff --git a/cmd/object-mutation-lease_test.go b/cmd/object-mutation-lease_test.go new file mode 100644 index 000000000..90e4b4004 --- /dev/null +++ b/cmd/object-mutation-lease_test.go @@ -0,0 +1,287 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/dsync" + "github.com/minio/minio/internal/grid" +) + +type scopeLeaseLocker struct { + *localLocker + partition *atomic.Bool + refresh chan<- struct{} +} + +func (l scopeLeaseLocker) Refresh(ctx context.Context, a dsync.LockArgs) (bool, error) { + if l.partition.Load() { + select { + case l.refresh <- struct{}{}: + default: + } + return false, errors.New("injected lock-plane partition") + } + return l.localLocker.Refresh(ctx, a) +} + +func (l scopeLeaseLocker) ForceUnlock(ctx context.Context, a dsync.LockArgs) (bool, error) { + if l.partition.Load() { + return false, errors.New("partition") + } + return l.localLocker.ForceUnlock(ctx, a) +} + +type scopeStagedContext struct { + context.Context + meta string + entered chan<- struct{} + release <-chan struct{} + once sync.Once + tracker healthDiskCtxValue +} + +func (c *scopeStagedContext) Value(k any) any { + if _, ok := k.(healthDiskCtxKey); ok { + if _, e := os.Stat(c.meta); e == nil { + c.once.Do(func() { c.entered <- struct{}{}; <-c.release }) + } + return &c.tracker + } + return c.Context.Value(k) +} + +type scopeRPCDisk struct { + StorageAPI + client *storageRESTClient +} + +func (d scopeRPCDisk) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, o RenameOptions) (RenameDataResp, error) { + return d.client.RenameData(ctx, sv, sp, fi, dv, dp, o) +} + +// Controlled fault schedule: real dsync lockers lose their expired lease, +// while a real disk RenameData is paused after reading and staging xl.meta. +// Only the clock age of lease records and the disk completion schedule are injected. +// No object data or metadata is rewritten by the test after the seed PUT. +func TestMutationLeaseLossAndPartition(t *testing.T) { + for _, partitioned := range []bool{false, true} { + t.Run(fmt.Sprintf("network-partition-%t", partitioned), func(t *testing.T) { + size, _ := strconv.Atoi(os.Getenv("SILO_TEST_MUTATION_BYTES")) + payload := func(s string) string { + if size <= len(s) { + return s + } + return s + strings.Repeat(".", size-len(s)) + } + z, bucket := mutationPools(t) + const object = "lease-key" + base := time.Now().Add(-time.Hour) + mutationPut(t, z, bucket, object, 0, payload("initial-0"), base) + var partition atomic.Bool + refreshed := make(chan struct{}, 16) + locals := make([]*localLocker, 4) + oldClients := make([]dsync.NetLocker, 4) + newClients := make([]dsync.NetLocker, 4) + for i := range locals { + locals[i] = newLocker() + oldClients[i] = scopeLeaseLocker{locals[i], &partition, refreshed} + newClients[i] = locals[i] + } + mklock := func(clients []dsync.NetLocker) *distLockInstance { + return &distLockInstance{rwMutex: dsync.NewDRWMutex(&dsync.Dsync{GetLockers: func() ([]dsync.NetLocker, string) { return clients, "scope-owner" }, Timeouts: dsync.DefaultTimeouts}, bucket+"/"+object), opsID: mustGetUUID()} + } + a := mklock(oldClients) + requestCtx, cancelRequest := context.WithCancel(context.Background()) + defer cancelRequest() + alc, err := a.GetLock(requestCtx, newDynamicTimeout(time.Second, time.Second)) + if err != nil { + t.Fatal(err) + } + defer a.Unlock(alc) + tg, err := grid.SetupTestGrid(2) + if err != nil { + t.Fatal(err) + } + defer tg.Cleanup() + entered, release := make(chan struct{}, 8), make(chan struct{}) + physicalDone := make(chan error, 8) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + set := z.serverPools[0].getHashedSet(object) + disks, swap := installListingDiskSwitch(set) + defer swap(disks) + wrapped := append([]StorageAPI(nil), disks...) + for i, disk := range disks { + route := fmt.Sprintf("/lease/%d", i) + handler := func(p *RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) { + var success int64 + gate := &scopeStagedContext{Context: context.Background(), meta: filepath.Join(disk.Endpoint().Path, p.SrcVolume, p.SrcPath, xlStorageFormatFile), entered: entered, release: release, tracker: healthDiskCtxValue{lastSuccess: &success}} + resp, e := disk.RenameData(gate, p.SrcVolume, p.SrcPath, p.FI, p.DstVolume, p.DstPath, p.Opts) + physicalDone <- e + return &resp, grid.NewRemoteErr(e) + } + if e := storageRenameDataRPC.Register(tg.Managers[1], handler, route); e != nil { + t.Fatal(e) + } + if e := storageRenameDataInlineRPC.Register(tg.Managers[1], func(p *RenameDataInlineHandlerParams) (*RenameDataResp, *grid.RemoteErr) { + defer p.Recycle() + return handler(&p.RenameDataHandlerParams) + }, route); e != nil { + t.Fatal(e) + } + client := &storageRESTClient{gridConn: tg.Managers[0].Connection(tg.Hosts[1]).Subroute(route)} + client.diskID.Store(new(string)) + wrapped[i] = scopeRPCDisk{disk, client} + } + swap(wrapped) + done := make(chan error, 1) + go func() { + body := []byte(payload("pending-1000")) + _, e := z.serverPools[0].PutObject(alc.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{NoLock: true, MTime: base.Add(time.Minute)}) + done <- e + }() + for range disks { + select { + case <-entered: + case e := <-done: + t.Fatalf("A returned before disk barrier: %v", e) + case <-time.After(5 * time.Second): + t.Fatal("disk barrier timeout") + } + } + t.Log("OBS A staged old metadata on all eight real disks; no final rename yet") + partition.Store(partitioned) + for _, l := range locals { + l.mutex.Lock() + for key, entries := range l.lockMap { + for i := range entries { + entries[i].TimeLastRefresh = time.Now().Add(-lockValidityDuration - time.Second).UnixNano() + } + l.lockMap[key] = entries + } + l.mutex.Unlock() + l.expireOldLocks(lockValidityDuration) + } + if partitioned { + select { + case <-refreshed: + case <-time.After(13 * time.Second): + t.Fatal("refresh not attempted") + } + time.Sleep(20 * time.Millisecond) + t.Logf("OBS after failed refresh old context error=%v", alc.Context().Err()) + } else { + select { + case <-alc.Context().Done(): + case <-time.After(13 * time.Second): + t.Fatal("confirmed lost lock did not cancel") + } + t.Logf("OBS confirmed lease loss canceled old context: %v", alc.Context().Err()) + } + b := mklock(newClients) + blc, e := b.GetLock(context.Background(), newDynamicTimeout(time.Second, time.Second)) + if e != nil { + t.Fatal(e) + } + swap(disks) + bdone := make(chan error, 1) + go func() { + body := []byte(payload("ack-9999")) + _, e := z.serverPools[0].PutObject(blc.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{NoLock: true, MTime: base.Add(2 * time.Minute)}) + b.Unlock(blc) + bdone <- e + }() + bearly := false + select { + case e := <-bdone: + if e != nil { + t.Fatal(e) + } + bearly = true + t.Log("OBS B acquired the same lock and ACKed while A remained physically staged") + case <-time.After(150 * time.Millisecond): + t.Log("OBS B is waiting for the earlier disk transaction") + } + if partitioned { + cancelRequest() + } + unblock() + select { + case e := <-done: + t.Logf("OBS A finished: %v", e) + case <-time.After(5 * time.Second): + t.Fatal("A did not finish") + } + if !bearly { + select { + case e := <-bdone: + if e != nil { + t.Fatal(e) + } + t.Log("OBS B ACKed after disk serialization") + case <-time.After(5 * time.Second): + t.Fatal("B did not finish") + } + } + + for range disks { + select { + case e := <-physicalDone: + t.Logf("OBS physical handler completed: %v", e) + case <-time.After(5 * time.Second): + t.Fatal("physical handler did not finish") + } + } + for _, fast := range []bool{false, true} { + r, e := z.GetObjectNInfo(context.Background(), bucket, object, nil, nil, ObjectOptions{FastGetObjInfo: fast}) + if e != nil { + t.Errorf("OBS latest acknowledged payload unreadable fast=%t error=%v", fast, e) + continue + } + data, e := io.ReadAll(r) + r.Close() + if e != nil { + t.Fatal(e) + } + shown := string(data) + if len(shown) > 32 { + shown = shown[:32] + } + t.Logf("OBS final GET fast=%t bytes=%d prefix=%q", fast, len(data), shown) + if string(data) != payload("ack-9999") && string(data) != payload("pending-1000") { + t.Errorf("STRICT STALE VALUE: prefix %q is neither latest ACK nor pending write", shown) + } + } + }) + } +} diff --git a/cmd/object-mutation.go b/cmd/object-mutation.go new file mode 100644 index 000000000..6d461c15d --- /dev/null +++ b/cmd/object-mutation.go @@ -0,0 +1,204 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "errors" + "maps" + "slices" + "strings" + "sync" +) + +// This is an undo capability, not a distributed fencing token. Any subsequent +// mutation of the same version revokes it, including a no-op metadata update. +const rollbackIDKey = ReservedMetadataPrefixLower + "rollback-id" + +// Bound memory independently of object count. Include the physical drive so +// shards of one object are not serialized across drives. Multiple paths are +// locked in slot order, including collisions, to avoid lock inversion. +var storageMutationLocks [65536]sync.Mutex + +func (s *xlStorage) lockMutations(paths ...string) func() { + var slots [4]int + n := 0 + for _, path := range paths { + h := uint64(14695981039346656037) + for _, part := range []string{s.drivePath, path} { + for i := range part { + h = (h ^ uint64(part[i])) * 1099511628211 + } + h *= 1099511628211 + } + slot := int(h % uint64(len(storageMutationLocks))) + if !slices.Contains(slots[:n], slot) { + slots[n] = slot + n++ + } + } + slices.Sort(slots[:n]) + for _, slot := range slots[:n] { + storageMutationLocks[slot].Lock() + } + return func() { + for i := n - 1; i >= 0; i-- { + storageMutationLocks[slots[i]].Unlock() + } + } +} + +// Compare logical version state, excluding per-shard checksums/index, physical +// DataDir placement and properties of other versions in the same xl.meta. +// Different DataDir values remain readable for the existing rebalance recovery +// contract; rollback separately requires an exact DataDir match. +func sameFileInfoVersion(a, b FileInfo) bool { + if a.VersionID != b.VersionID || a.Deleted != b.Deleted || a.XLV1 != b.XLV1 || a.Size != b.Size || + a.TransitionStatus != b.TransitionStatus || a.TransitionTier != b.TransitionTier || + a.TransitionedObjName != b.TransitionedObjName || a.TransitionVersionID != b.TransitionVersionID || + !maps.Equal(a.Metadata, b.Metadata) || !bytes.Equal(a.Checksum, b.Checksum) || len(a.Parts) != len(b.Parts) { + return false + } + if !a.Deleted && a.Size != 0 && !a.Erasure.Equal(b.Erasure) { + return false + } + for i, p := range a.Parts { + q := b.Parts[i] + if p.Number != q.Number || p.Size != q.Size || p.ActualSize != q.ActualSize || p.ETag != q.ETag || !bytes.Equal(p.Index, q.Index) { + return false + } + } + return true +} + +// Legacy callers without an undo capability must match the persisted state, +// rather than transient coordinator-only fields in their original FileInfo. +func samePersistedVersion(current, expected FileInfo) bool { + var meta xlMetaV2 + if err := meta.AddVersion(expected); err != nil { + return false + } + fi, err := meta.ToFileInfo(current.Volume, current.Name, expected.VersionID, true, true) + return err == nil && sameFileInfoVersion(current, fi) +} + +// Caller holds the object's mutation guard. Restore just the overwritten +// version, and persist that decision before removing any external shards. +func (s *xlStorage) undoWrite(ctx context.Context, volume, path string, fi FileInfo, opts DeleteOptions, meta *xlMetaV2) error { + versionID := fi.VersionID + if versionID == "" { + versionID = nullVersionID + } + current, err := meta.ToFileInfo(volume, path, versionID, true, true) + if err != nil { + return err + } + if current.DataDir != fi.DataDir || !current.ModTime.Equal(fi.ModTime) || !samePersistedVersion(current, fi) { + return errFileVersionNotFound + } + var previous xlMetaV2 + var oldVersion *xlMetaV2Version + var oldData []byte + volumeDir, err := s.getVolDir(volume) + if err != nil { + return err + } + if opts.OldDataDir != "" { + buf, err := s.readAllData(ctx, volume, volumeDir, pathJoin(volumeDir, path, opts.OldDataDir, xlStorageFormatFileBackup)) + if err != nil { + return err + } + defer metaDataPoolPut(buf) + if err := previous.LoadOrConvert(buf); err != nil { + return err + } + _, oldVersion, err = previous.findVersionStr(versionID) + if err != nil { + return err + } + oldData = previous.data.find(versionID) + if oldData == nil { + oldData = previous.data.find(oldVersion.getDataDir()) + } + } + dataDir, err := meta.DeleteVersion(fi) + if err != nil { + return err + } + if dataDir != "" { + meta.data.remove(versionID, dataDir) + } + if oldVersion != nil { + if err := meta.addVersion(*oldVersion); err != nil { + return err + } + if oldData != nil { + meta.data.replace(versionID, oldData) + } + if dataDir == oldVersion.getDataDir() { + dataDir = "" + } + } + if len(meta.versions) > 0 { + buf, err := meta.AppendTo(metaDataPoolGet()) + defer metaDataPoolPut(buf) + if err != nil { + return err + } + if err := s.writeAllMeta(ctx, volume, pathJoin(path, xlStorageFormatFile), buf, true); err != nil { + return err + } + } else if err := s.deleteFile(volumeDir, pathJoin(volumeDir, path, xlStorageFormatFile), true, false); err != nil { + return err + } + if dataDir != "" { + if err := s.moveToTrash(pathJoin(volumeDir, path, dataDir), true, false); err != nil && !errors.Is(err, errFileNotFound) { + return err + } + } + if opts.OldDataDir != "" { + backupDir := pathJoin(volumeDir, path, opts.OldDataDir) + Remove(pathJoin(backupDir, xlStorageFormatFileBackup)) + if strings.HasSuffix(opts.OldDataDir, "-rollback") { + Remove(backupDir) + } + } + return nil +} + +// All drives have replied. Only a uniform indexed null version may avoid the +// second metadata decode. Histories and conflicts retain the full merge. Exact +// FileInfo comparison also prevents the four-byte header signature from being +// treated as a collision-free identity. +func uniformLatestVersion(raw []RawFileInfo, meta []FileInfo, errs []error) bool { + if len(raw) == 0 { + return false + } + var first xlMetaV2VersionHeader + for i, r := range raw { + if errs[i] != nil || !meta[i].IsValid() || !sameFileInfoVersion(meta[0], meta[i]) { + return false + } + buf, _, err := isIndexedMetaV2(r.Buf) + if err != nil || buf == nil { + return false + } + n, hv, mv, rest, err := decodeXLHeaders(buf) + if err != nil || n != 1 || mv != xlMetaVersion { + return false + } + var h xlMetaV2VersionHeader + err = decodeVersions(rest, n, func(_ int, hdr, _ []byte) error { _, e := h.unmarshalV(hv, hdr); return e }) + if err != nil || h.Type != ObjectType || h.FreeVersion() || h.VersionID != [16]byte{} { + return false + } + if i == 0 { + first = h + } else if h != first { + return false + } + } + return true +} diff --git a/cmd/rename-cancellation_test.go b/cmd/rename-cancellation_test.go index 63230a753..0c1a071dc 100644 --- a/cmd/rename-cancellation_test.go +++ b/cmd/rename-cancellation_test.go @@ -316,6 +316,8 @@ func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { unblock := func() { releaseOnce.Do(func() { close(release) }) } defer unblock() var completed atomic.Int32 + var injectCommitFailure atomic.Bool + injectCommitFailure.Store(true) for pi, pool := range z.serverPools { set := pool.getHashedSet(object) disks, switchDisks := installListingDiskSwitch(set) @@ -330,8 +332,14 @@ func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { handle := func(p *RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) { entered <- struct{}{} <-release + // Cancellation alone must not undo a successful quorum. + // This rollback fixture also fails half the disks, so its + // original payload/lock assertions require a real failed commit. + defer completed.Add(1) + if i >= len(disks)/2 && injectCommitFailure.Load() { + return nil, grid.NewRemoteErr(errDiskNotFound) + } resp, err := disk.RenameData(context.Background(), p.SrcVolume, p.SrcPath, p.FI, p.DstVolume, p.DstPath, p.Opts) - completed.Add(1) if err != nil { return nil, grid.NewRemoteErr(err) } @@ -427,6 +435,7 @@ func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { if completed.Load() != int32(n) { t.Fatal("returned without accounting for every remote commit") } + injectCommitFailure.Store(false) for pi, pool := range z.serverPools { for _, version := range []string{"", nullVersionID} { r, err := pool.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{VersionID: version}) diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index 77248f72a..42d683c37 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -1570,7 +1570,11 @@ func (x *xlMetaV2) UpdateObjectVersion(fi FileInfo) error { if err != nil { return err } + delete(ver.ObjectV2.MetaSys, rollbackIDKey) for k, v := range fi.Metadata { + if k == rollbackIDKey { + continue + } if len(k) > len(ReservedMetadataPrefixLower) && strings.EqualFold(k[:len(ReservedMetadataPrefixLower)], ReservedMetadataPrefixLower) { ver.ObjectV2.MetaSys[k] = []byte(v) } else { diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go index 2a224982c..822f797ca 100644 --- a/cmd/xl-storage.go +++ b/cmd/xl-storage.go @@ -25,6 +25,7 @@ import ( "errors" "fmt" "io" + "maps" "os" pathutil "path" "path/filepath" @@ -1100,6 +1101,7 @@ func (s *xlStorage) ListDir(ctx context.Context, origvolume, volume, dirPath str } func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis ...FileInfo) error { + defer s.lockMutations(pathJoin(volume, path))() volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -1308,11 +1310,10 @@ func (s *xlStorage) readAllData(ctx context.Context, volume, volumeDir string, f }) } -func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) (err error) { - w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout()) - return w.Run(func() (err error) { - return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge) - }) +func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) error { + // The caller must retain its mutation guard until this rename completes. + // Only purging the already detached unique trash directory is asynchronous. + return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge) } // DeleteVersion - deletes FileInfo metadata for path at `xl.meta`. forceDelMarker @@ -1325,6 +1326,8 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F }) } + defer s.lockMutations(pathJoin(volume, path))() + volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -1345,7 +1348,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F metaDataPoolPut(buf) // Never used, return it if fi.Deleted && forceDelMarker { // Create a new xl.meta with a delete marker in it - return s.WriteMetadata(ctx, "", volume, path, fi) + return s.writeMetadataNoLock(ctx, "", volume, path, fi) } s.RLock() @@ -1382,19 +1385,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F } if opts.UndoWrite { - versionID := fi.VersionID - if versionID == "" { - versionID = nullVersionID - } - current, err := xlMeta.ToFileInfo(volume, path, versionID, true, false) - if err != nil { - return err - } - // A rollback only owns the generation it committed. A delayed undo - // must never delete a subsequent write to the same null/version ID. - if current.DataDir != fi.DataDir || !current.ModTime.Equal(fi.ModTime) { - return errFileVersionNotFound - } + return s.undoWrite(ctx, volume, path, fi, opts, &xlMeta) } dataDir, err := xlMeta.DeleteVersion(fi) @@ -1426,12 +1417,6 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F } } - // Restore the complete prior metadata even if other versions remain. - // Deleting just the failed version would lose the null version it replaced. - if opts.UndoWrite && opts.OldDataDir != "" { - return renameAll(pathJoin(filePath, opts.OldDataDir, xlStorageFormatFileBackup), pathJoin(filePath, xlStorageFormatFile), filePath) - } - if len(xlMeta.versions) != 0 { // xl.meta must still exist for other versions, dataDir is purged. buf, err = xlMeta.AppendTo(metaDataPoolGet()) @@ -1448,6 +1433,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F // Updates only metadata for a given version. func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) (err error) { + defer s.lockMutations(pathJoin(volume, path))() if len(fi.Metadata) == 0 { return errInvalidArgument } @@ -1496,6 +1482,16 @@ func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi // WriteMetadata - writes FileInfo metadata for path at `xl.meta` func (s *xlStorage) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) { + defer s.lockMutations(pathJoin(volume, path))() + return s.writeMetadataNoLock(ctx, origvolume, volume, path, fi) +} + +func (s *xlStorage) writeMetadataNoLock(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) { + // Any later write of this version revokes the original writer's undo. + if _, ok := fi.Metadata[rollbackIDKey]; ok { + fi.Metadata = maps.Clone(fi.Metadata) + delete(fi.Metadata, rollbackIDKey) + } if fi.Fresh { if origvolume != "" { origvolumeDir, err := s.getVolDir(origvolume) @@ -2558,7 +2554,10 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri return err } - if err = s.moveToTrash(filePath, false, false); err != nil { + unlock := s.lockMutations(pathJoin(volume, fp), pathJoin(volume, pathutil.Dir(fp))) + err = s.moveToTrash(filePath, false, false) + unlock() + if err != nil { return err } } @@ -2568,6 +2567,7 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri // Delete - delete a file at path. func (s *xlStorage) Delete(ctx context.Context, volume string, path string, deleteOpts DeleteOptions) (err error) { + defer s.lockMutations(pathJoin(volume, path), pathJoin(volume, pathutil.Dir(path)))() volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -2597,6 +2597,7 @@ func skipAccessChecks(volume string) (ok bool) { // RenameData - rename source path to destination path atomically, metadata and data directory. func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (res RenameDataResp, err error) { + defer s.lockMutations(pathJoin(srcVolume, srcPath), pathJoin(dstVolume, dstPath))() defer func() { ignoredErrs := []error{ errFileNotFound,