diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 5c9ee5d80..43f9c54b2 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -659,6 +659,17 @@ func pickLatestQuorumFilesInfo(ctx context.Context, rawFileInfos []RawFileInfo,
for i := range errs {
if errs[i] == nil {
errs[i] = err
+ // An empty quorum merge does not prove absence when this
+ // disk contains a readable object generation. Preserve that
+ // uncertainty instead of hiding it behind FileNotFound.
+ if errors.Is(err, errFileNotFound) && metadataArray[i] != nil {
+ for _, ver := range metadataArray[i].versions {
+ if inclFreeVers || !ver.header.FreeVersion() {
+ errs[i] = errErasureReadQuorum
+ break
+ }
+ }
+ }
}
}
return metaFileInfos, errs
@@ -1000,16 +1011,6 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
}
}
}
- // when we have insufficient read quorum and inconsistent metadata return
- // file not found, since we can't possibly have a way to recover this object
- // anyway.
- if v, ok := err.(InsufficientReadQuorum); ok && v.Type == RQInconsistentMeta {
- if opts.VersionID != "" {
- err = errFileVersionNotFound
- } else {
- err = errFileNotFound
- }
- }
return fi, nil, nil, toObjectErr(err, bucket, object)
}
diff --git a/cmd/erasure-read-generation_test.go b/cmd/erasure-read-generation_test.go
new file mode 100644
index 000000000..bdb9c2c08
--- /dev/null
+++ b/cmd/erasure-read-generation_test.go
@@ -0,0 +1,125 @@
+// 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"
+ "fmt"
+ "net/http"
+ "testing"
+ "time"
+)
+
+// All metadata is valid and readable. Only the completed generations differ.
+// No timing, missing files, synthetic read error or malformed metadata is needed.
+func TestReadGenerationConflictIsNotAbsence(t *testing.T) {
+ raw := make([]RawFileInfo, 8)
+ for i := range raw {
+ fi := newFileInfo("object", 4, 4)
+ fi.Erasure.Index = i + 1
+ fi.ModTime = time.Unix(100+int64(i%4), 0)
+ fi.DataDir = mustGetUUID()
+ fi.Metadata = map[string]string{"etag": fmt.Sprint(i % 4)}
+ var meta xlMetaV2
+ if err := meta.AddVersion(fi); err != nil {
+ t.Fatal(err)
+ }
+ buf, err := meta.AppendTo(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw[i] = RawFileInfo{Buf: buf}
+ }
+ _, errs := pickLatestQuorumFilesInfo(t.Context(), raw, make([]error, len(raw)), "bucket", "object", true, false)
+ err := reduceReadQuorumErrs(t.Context(), errs, objectOpIgnoredErrs, 4)
+ if api := toAPIError(t.Context(), toObjectErr(err, "bucket", "object")); api.HTTPStatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("readable conflicting generations reported as absence: %v (%+v)", err, api)
+ }
+}
+
+func TestPoolsReadGenerationConflictDoesNotExposeOldValue(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ router := initTestAPIEndPoints(z, []string{"GetObject", "HeadObject"})
+ for broken := range 2 {
+ t.Run(fmt.Sprintf("broken-pool-%d", broken), func(t *testing.T) {
+ object := fmt.Sprintf("conflict-%d", broken)
+ base := UTCNow().Add(-time.Hour)
+ putConsistencyObject(t, z, bucket, object, 1-broken, "old-acknowledged", ObjectOptions{MTime: base})
+ set := z.serverPools[broken].getHashedSet(object)
+ disks := set.getDisks()
+ saved := make([][]byte, len(disks))
+ for generation := range 4 {
+ putConsistencyObject(t, z, bucket, object, broken, fmt.Sprintf("newer-%d", generation),
+ ObjectOptions{MTime: base.Add(time.Duration(generation+1) * time.Minute)})
+ for i, disk := range disks {
+ if i%4 != generation {
+ continue
+ }
+ buf, err := disk.ReadAll(ctx, bucket, pathJoin(object, xlStorageFormatFile))
+ if err != nil {
+ t.Fatal(err)
+ }
+ saved[i] = bytes.Clone(buf)
+ }
+ }
+ for i, disk := range disks {
+ if err := disk.WriteAll(ctx, bucket, pathJoin(object, xlStorageFormatFile), saved[i]); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, version := range []string{"", nullVersionID} {
+ for _, method := range []string{http.MethodHead, http.MethodGet} {
+ rec := consistencyRequest(t, router, method, bucket, object, version)
+ if rec.Code != http.StatusServiceUnavailable {
+ t.Errorf("%s version=%q exposed an old value or false absence: status=%d body=%q", method, version, rec.Code, rec.Body.String())
+ }
+ }
+ }
+ 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)
+ }
+ }
+ })
+ }
+}
+
+func TestPoolsReadUnavailablePoolDoesNotExposeOldValue(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ for broken := range 2 {
+ t.Run(fmt.Sprintf("broken-pool-%d", broken), func(t *testing.T) {
+ object := fmt.Sprintf("unavailable-%d", broken)
+ putConsistencyObject(t, z, bucket, object, 1-broken, "old-acknowledged", ObjectOptions{})
+ set := z.serverPools[broken].getHashedSet(object)
+ disks, switchDisks := installListingDiskSwitch(set)
+ defer switchDisks(disks)
+ faulty := append([]StorageAPI(nil), disks...)
+ for i := range faulty {
+ faulty[i] = consistencyReadFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object}
+ }
+ switchDisks(faulty)
+ if _, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); err == nil || isErrObjectNotFound(err) {
+ t.Errorf("unknown pool state allowed an old value: %v", err)
+ }
+ })
+ }
+}
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index a75af7339..d1b154edb 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1048,6 +1048,17 @@ func (z *erasureServerPools) getLatestObjectInfoWithIdx(ctx context.Context, buc
}
wg.Wait()
+ // An unreadable pool may contain a newer generation. Check every pool
+ // before choosing a readable copy; sorting first would hide errors behind
+ // an older object's nonzero ModTime. Delete markers are resolved metadata
+ // and still participate in the ordering below.
+ for _, res := range results {
+ if res.err != nil && !isErrObjectNotFound(res.err) && !isErrVersionNotFound(res.err) &&
+ (!res.oi.DeleteMarker || !isErrMethodNotAllowed(res.err)) {
+ return res.oi, res.zIdx, res.err
+ }
+ }
+
// Sort the objInfos such that we always serve latest
// this is a defensive change to handle any duplicate
// content that may have been created, we always serve
diff --git a/cmd/rename-cancellation_test.go b/cmd/rename-cancellation_test.go
new file mode 100644
index 000000000..1f78f8d87
--- /dev/null
+++ b/cmd/rename-cancellation_test.go
@@ -0,0 +1,454 @@
+// 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"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/minio/minio/internal/grid"
+)
+
+// Pause at the real disk health callback after staging xl.meta. Unlike a sleep,
+// this fixes cancellation between staging and the final metadata rename.
+type stagedRenameContext struct {
+ context.Context
+ meta string
+ entered, release chan struct{}
+ once sync.Once
+ tracker healthDiskCtxValue
+}
+
+func (c *stagedRenameContext) Value(key any) any {
+ if _, ok := key.(healthDiskCtxKey); ok {
+ if _, err := os.Stat(c.meta); err == nil {
+ c.once.Do(func() {
+ close(c.entered)
+ <-c.release
+ })
+ }
+ return &c.tracker
+ }
+ return c.Context.Value(key)
+}
+
+func TestRenameCancellationLocalWaitsForDisk(t *testing.T) {
+ disk, root, err := newXLStorageTestSetup(t)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer disk.Close()
+ if err := disk.MakeVol(t.Context(), "bucket"); err != nil {
+ t.Fatal(err)
+ }
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ var lastSuccess int64
+ gate := &stagedRenameContext{
+ Context: ctx, meta: filepath.Join(root, minioMetaTmpBucket, "staged", xlStorageFormatFile),
+ entered: make(chan struct{}), release: make(chan struct{}),
+ tracker: healthDiskCtxValue{lastSuccess: &lastSuccess},
+ }
+ fi := newFileInfo("object", 2, 2)
+ fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1
+ fi.Fresh = true
+ done := make(chan error, 1)
+ go func() {
+ _, err := disk.RenameData(gate, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{})
+ done <- err
+ }()
+ <-gate.entered
+ cancel()
+ synctest.Wait()
+ select {
+ case err := <-done:
+ t.Errorf("RenameData returned %v while its disk worker was still running", err)
+ default:
+ }
+ close(gate.release)
+ synctest.Wait()
+ select {
+ case err := <-done:
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("canceled staged write: %v", err)
+ }
+ default:
+ }
+ if _, err := os.Stat(filepath.Join(root, "bucket", "object", xlStorageFormatFile)); !os.IsNotExist(err) {
+ t.Fatalf("canceled staged write committed metadata: %v", err)
+ }
+ })
+}
+
+func TestRenameCancellationRPCWaitsForDisk(t *testing.T) {
+ for _, operation := range []string{"regular", "inline", "write-metadata", "update-metadata", "undo"} {
+ t.Run(operation, func(t *testing.T) {
+ tg, err := grid.SetupTestGrid(2)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tg.Cleanup()
+ entered, release := make(chan struct{}), make(chan struct{})
+ var committed atomic.Bool
+ handle := func(*RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) {
+ close(entered)
+ <-release
+ committed.Store(true)
+ return &RenameDataResp{OldDataDir: "previous-generation"}, nil
+ }
+ const route = "/cancel-commit"
+ if err := storageRenameDataRPC.Register(tg.Managers[1], handle, route); err != nil {
+ t.Fatal(err)
+ }
+ if err := storageRenameDataInlineRPC.Register(tg.Managers[1], func(p *RenameDataInlineHandlerParams) (*RenameDataResp, *grid.RemoteErr) {
+ defer p.Recycle()
+ return handle(&p.RenameDataHandlerParams)
+ }, route); err != nil {
+ t.Fatal(err)
+ }
+ meta := func(*MetadataHandlerParams) (grid.NoPayload, *grid.RemoteErr) {
+ _, err := handle(nil)
+ return grid.NewNoPayload(), err
+ }
+ if err := storageWriteMetadataRPC.Register(tg.Managers[1], meta, route); err != nil {
+ t.Fatal(err)
+ }
+ if err := storageUpdateMetadataRPC.Register(tg.Managers[1], meta, route); err != nil {
+ t.Fatal(err)
+ }
+ if err := storageDeleteVersionRPC.Register(tg.Managers[1], func(*DeleteVersionHandlerParams) (grid.NoPayload, *grid.RemoteErr) {
+ return meta(nil)
+ }, route); err != nil {
+ t.Fatal(err)
+ }
+ client := &storageRESTClient{gridConn: tg.Managers[0].Connection(tg.Hosts[1]).Subroute(route)}
+ client.diskID.Store(new(string))
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ fi := FileInfo{}
+ if operation == "inline" {
+ fi.Data = []byte("payload")
+ }
+ type result struct {
+ resp RenameDataResp
+ err error
+ }
+ done := make(chan result, 1)
+ go func() {
+ var resp RenameDataResp
+ var err error
+ switch operation {
+ case "write-metadata":
+ err = client.WriteMetadata(ctx, "", "bucket", "object", fi)
+ case "update-metadata":
+ err = client.UpdateMetadata(ctx, "bucket", "object", fi, UpdateMetadataOpts{})
+ case "undo":
+ err = client.DeleteVersion(ctx, "bucket", "object", fi, false, DeleteOptions{UndoWrite: true})
+ default:
+ resp, err = client.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{})
+ }
+ done <- result{resp, err}
+ }()
+ <-entered
+ cancel()
+ var early *result
+ select {
+ case r := <-done:
+ early = &r
+ t.Errorf("RPC returned %v before the remote disk completed", r.err)
+ case <-time.After(100 * time.Millisecond):
+ }
+ close(release)
+ if early == nil {
+ select {
+ case r := <-done:
+ wantOld := ""
+ if operation == "regular" || operation == "inline" {
+ wantOld = "previous-generation"
+ }
+ if r.err != nil || r.resp.OldDataDir != wantOld || !committed.Load() {
+ t.Errorf("lost completed result needed for rollback: %+v", r)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("RPC did not finish after disk completion")
+ }
+ }
+ })
+ }
+}
+
+func TestRenameCancellationAlreadyCanceledRPC(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ // A canceled call must not access the connection or send a new mutation.
+ client := &storageRESTClient{}
+ client.diskID.Store(new(string))
+ defer func() {
+ if p := recover(); p != nil {
+ t.Errorf("canceled mutation reached the connection: %v", p)
+ }
+ }()
+ _, err := client.RenameData(ctx, minioMetaTmpBucket, "staged", FileInfo{}, "bucket", "object", RenameOptions{})
+ if !errors.Is(err, context.Canceled) {
+ t.Fatalf("pre-canceled RenameData: %v", err)
+ }
+}
+
+type gatedRenameDisk struct {
+ StorageAPI
+ client *storageRESTClient
+}
+
+func (d gatedRenameDisk) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, opts RenameOptions) (RenameDataResp, error) {
+ return d.client.RenameData(ctx, sv, sp, fi, dv, dp, opts)
+}
+
+func (d gatedRenameDisk) WriteMetadata(ctx context.Context, ov, volume, path string, fi FileInfo) error {
+ return d.client.WriteMetadata(ctx, ov, volume, path, fi)
+}
+
+// Use real disk commits behind the real RPC client. The barrier fixes the
+// order: every commit has started, then cancellation, then remote completion.
+func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) {
+ for _, tc := range []struct {
+ size int
+ versioned bool
+ copy bool
+ }{
+ {32, false, false},
+ {2 << 20, false, false},
+ {32, true, false},
+ {2 << 20, true, false},
+ {32, false, true},
+ {2 << 20, false, true},
+ } {
+ t.Run(fmt.Sprintf("bytes-%d/other-versions-%t/copy-%t", tc.size, tc.versioned, tc.copy), func(t *testing.T) {
+ size := tc.size
+ z, bucket := consistencyPools(t)
+ object := "cancel-rollback"
+ old := strings.Repeat("a", size)
+ base := UTCNow().Add(-time.Hour)
+ for pool := range 2 {
+ if tc.versioned {
+ putConsistencyObject(t, z, bucket, object, pool, "retained-version", ObjectOptions{Versioned: true, MTime: base.Add(-time.Minute)})
+ }
+ putConsistencyObject(t, z, bucket, object, pool, old, ObjectOptions{MTime: base})
+ }
+ copyInfo, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ copyInfo.metadataOnly = true
+ copyInfo.UserDefined["copy-test"] = "canceled"
+ tg, err := grid.SetupTestGrid(2)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tg.Cleanup()
+ entered := make(chan struct{}, 32)
+ release := make(chan struct{})
+ var releaseOnce sync.Once
+ unblock := func() { releaseOnce.Do(func() { close(release) }) }
+ defer unblock()
+ var completed atomic.Int32
+ for pi, pool := range z.serverPools {
+ set := pool.getHashedSet(object)
+ disks, switchDisks := installListingDiskSwitch(set)
+ defer switchDisks(disks)
+ wrapped := append([]StorageAPI(nil), disks...)
+ for i, disk := range disks {
+ route := fmt.Sprintf("/cancel-rollback/%d/%d", pi, i)
+ handle := func(p *RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) {
+ entered <- struct{}{}
+ <-release
+ 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)
+ }
+ return &resp, nil
+ }
+ if err := storageRenameDataRPC.Register(tg.Managers[1], handle, route); err != nil {
+ t.Fatal(err)
+ }
+ if err := storageRenameDataInlineRPC.Register(tg.Managers[1], func(p *RenameDataInlineHandlerParams) (*RenameDataResp, *grid.RemoteErr) {
+ defer p.Recycle()
+ return handle(&p.RenameDataHandlerParams)
+ }, route); err != nil {
+ t.Fatal(err)
+ }
+ if err := storageWriteMetadataRPC.Register(tg.Managers[1], func(p *MetadataHandlerParams) (grid.NoPayload, *grid.RemoteErr) {
+ if p.Volume == bucket && p.FilePath == object {
+ entered <- struct{}{}
+ <-release
+ defer completed.Add(1)
+ }
+ err := disk.WriteMetadata(context.Background(), p.OrigVolume, p.Volume, p.FilePath, p.FI)
+ return grid.NewNoPayload(), grid.NewRemoteErr(err)
+ }, route); err != nil {
+ t.Fatal(err)
+ }
+ client := &storageRESTClient{gridConn: tg.Managers[0].Connection(tg.Hosts[1]).Subroute(route)}
+ client.diskID.Store(new(string))
+ wrapped[i] = gatedRenameDisk{disk, client}
+ }
+ switchDisks(wrapped)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ pending := bytes.Repeat([]byte("b"), size)
+ reader := mustGetPutObjReader(t, bytes.NewReader(pending), int64(size), "", "")
+ done := make(chan error, 1)
+ go func() {
+ var err error
+ if tc.copy {
+ _, err = z.CopyObject(ctx, bucket, object, bucket, object, copyInfo, ObjectOptions{}, ObjectOptions{})
+ } else {
+ _, err = z.PutObject(ctx, bucket, object, reader, ObjectOptions{})
+ }
+ done <- err
+ }()
+ n := z.serverPools[0].getHashedSet(object).setDriveCount
+ for range n {
+ select {
+ case <-entered:
+ case err := <-done:
+ t.Fatalf("write ended before commit barrier: %v", err)
+ case <-time.After(5 * time.Second):
+ t.Fatal("write did not reach every disk")
+ }
+ }
+ cancel()
+ var early bool
+ select {
+ case err := <-done:
+ early = true
+ t.Errorf("mutation returned before remote commits completed: %v", err)
+ case <-time.After(100 * time.Millisecond):
+ }
+ lock := z.NewNSLock(bucket, object)
+ lockCtx, cancelLock := context.WithTimeout(t.Context(), 50*time.Millisecond)
+ lc, lockErr := lock.GetLock(lockCtx, newDynamicTimeout(50*time.Millisecond, 50*time.Millisecond))
+ cancelLock()
+ if lockErr == nil {
+ lock.Unlock(lc)
+ t.Error("namespace lock released before remote commits and rollback")
+ }
+ unblock()
+ if early {
+ // Keep the old-code failure deterministic; don't race cleanup
+ // against handlers intentionally still running at the barrier.
+ deadline := time.Now().Add(5 * time.Second)
+ for completed.Load() != int32(n) {
+ if time.Now().After(deadline) {
+ t.Fatal("remote handlers did not finish")
+ }
+ time.Sleep(time.Millisecond)
+ }
+ return
+ }
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("canceled write unexpectedly acknowledged")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("canceled write did not finish rollback")
+ }
+ if completed.Load() != int32(n) {
+ t.Fatal("returned without accounting for every remote commit")
+ }
+ for pi, pool := range z.serverPools {
+ r, err := pool.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatalf("pool %d lost its acknowledged copy: %v", pi, err)
+ }
+ body, err := io.ReadAll(r)
+ r.Close()
+ if err != nil || string(body) != old {
+ t.Errorf("pool %d rollback lost acknowledged payload: bytes=%d err=%v", pi, len(body), err)
+ }
+ }
+ got, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(got)
+ got.Close()
+ if err != nil || string(body) != old {
+ t.Fatalf("rollback lost acknowledged payload: bytes=%d err=%v", len(body), err)
+ }
+ newBody := bytes.Repeat([]byte("c"), size)
+ if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(size), "", ""), ObjectOptions{}); err != nil {
+ t.Fatal(err)
+ }
+ got, err = z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err = io.ReadAll(got)
+ got.Close()
+ if err != nil || !bytes.Equal(body, newBody) {
+ t.Fatalf("late commit replaced subsequent successful PUT: bytes=%d err=%v", len(body), err)
+ }
+ })
+ }
+}
+
+func TestRenameCancellationStaleRollbackPreservesNewWrite(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ const object = "stale-rollback"
+ putConsistencyObject(t, z, bucket, object, 0, "canceled-generation", ObjectOptions{})
+ disks := z.serverPools[0].getHashedSet(object).getDisks()
+ stale := make([]FileInfo, len(disks))
+ for i, disk := range disks {
+ var err error
+ stale[i], err = disk.ReadVersion(t.Context(), "", bucket, object, nullVersionID, ReadOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ putConsistencyObject(t, z, bucket, object, 0, "new-acknowledged-generation", ObjectOptions{})
+ for i, disk := range disks {
+ err := disk.DeleteVersion(t.Context(), bucket, object, stale[i], false, DeleteOptions{UndoWrite: true})
+ if err != nil && !errors.Is(err, errFileVersionNotFound) {
+ t.Fatal(err)
+ }
+ }
+ got, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatalf("stale rollback removed the new acknowledged write: %v", err)
+ }
+ defer got.Close()
+ body, err := io.ReadAll(got)
+ if err != nil || string(body) != "new-acknowledged-generation" {
+ t.Fatalf("stale rollback changed the new acknowledged payload: %q, %v", body, err)
+ }
+}
diff --git a/cmd/storage-rest-client.go b/cmd/storage-rest-client.go
index 0ba573847..302858a41 100644
--- a/cmd/storage-rest-client.go
+++ b/cmd/storage-rest-client.go
@@ -408,8 +408,11 @@ func (client *storageRESTClient) CreateFile(ctx context.Context, origvolume, vol
}
func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error {
- ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
- defer cancel()
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ // As with RenameData, admitted metadata writes must finish before unlock.
+ ctx = context.WithoutCancel(ctx)
_, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
DiskID: *client.diskID.Load(),
@@ -422,8 +425,10 @@ func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume,
}
func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error {
- ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
- defer cancel()
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ ctx = context.WithoutCancel(ctx)
_, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
DiskID: *client.diskID.Load(),
@@ -436,8 +441,14 @@ func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, pat
}
func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) (err error) {
- ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
- defer cancel()
+ if opts.UndoWrite {
+ // Rollback is part of the commit and must finish before unlocking.
+ ctx = context.WithoutCancel(ctx)
+ } else {
+ var cancel context.CancelFunc
+ ctx, cancel = context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
+ defer cancel()
+ }
_, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{
DiskID: *client.diskID.Load(),
@@ -487,6 +498,13 @@ func (client *storageRESTClient) CheckParts(ctx context.Context, volume string,
func (client *storageRESTClient) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo,
dstVolume, dstPath string, opts RenameOptions,
) (res RenameDataResp, err error) {
+ if err := ctx.Err(); err != nil {
+ return res, err
+ }
+ // The remote handler cannot cancel an admitted RenameData. Keep waiting
+ // for its result so renameData can roll back every completed mutation
+ // before the caller cleans staging files and releases the namespace lock.
+ ctx = context.WithoutCancel(ctx)
params := RenameDataHandlerParams{
DiskID: *client.diskID.Load(),
SrcVolume: srcVolume,
diff --git a/cmd/xl-storage-disk-id-check.go b/cmd/xl-storage-disk-id-check.go
index 39fe68ff1..408981c2f 100644
--- a/cmd/xl-storage-disk-id-check.go
+++ b/cmd/xl-storage-disk-id-check.go
@@ -495,7 +495,7 @@ func (p *xlStorageDiskIDCheck) RenameData(ctx context.Context, srcVolume, srcPat
if len(fi.Data) > 0 {
fi.Data = append(grid.GetByteBufferCap(len(fi.Data))[:0], fi.Data...)
}
- return xioutil.WithDeadline[RenameDataResp](ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (res RenameDataResp, err error) {
+ return xioutil.WithDeadlineAndWait[RenameDataResp](ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (res RenameDataResp, err error) {
if len(fi.Data) > 0 {
defer grid.PutByteBuffer(fi.Data)
}
@@ -634,6 +634,12 @@ func (p *xlStorageDiskIDCheck) DeleteVersion(ctx context.Context, volume, path s
}
}()
+ if opts.UndoWrite {
+ _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) {
+ return struct{}{}, p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts)
+ })
+ return err
+ }
w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
return w.Run(func() error { return p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) })
}
@@ -645,8 +651,10 @@ func (p *xlStorageDiskIDCheck) UpdateMetadata(ctx context.Context, volume, path
}
defer done(0, &err)
- w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
- return w.Run(func() error { return p.storage.UpdateMetadata(ctx, volume, path, fi, opts) })
+ _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) {
+ return struct{}{}, p.storage.UpdateMetadata(ctx, volume, path, fi, opts)
+ })
+ return err
}
func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) {
@@ -656,8 +664,10 @@ func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, origvolume, vo
}
defer done(0, &err)
- w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
- return w.Run(func() error { return p.storage.WriteMetadata(ctx, origvolume, volume, path, fi) })
+ _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) {
+ return struct{}{}, p.storage.WriteMetadata(ctx, origvolume, volume, path, fi)
+ })
+ return err
}
func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) {
diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go
index 81e415553..426444fc2 100644
--- a/cmd/xl-storage.go
+++ b/cmd/xl-storage.go
@@ -1381,6 +1381,22 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
return err
}
+ 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
+ }
+ }
+
dataDir, err := xlMeta.DeleteVersion(fi)
if err != nil {
return err
@@ -1410,6 +1426,12 @@ 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())
@@ -1421,10 +1443,6 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
return s.writeAllMeta(ctx, volume, pathJoin(path, xlStorageFormatFile), buf, true)
}
- if opts.UndoWrite && opts.OldDataDir != "" {
- return renameAll(pathJoin(filePath, opts.OldDataDir, xlStorageFormatFileBackup), pathJoin(filePath, xlStorageFormatFile), filePath)
- }
-
return s.deleteFile(volumeDir, pathJoin(volumeDir, path, xlStorageFormatFile), true, false)
}
diff --git a/internal/ioutil/ioutil.go b/internal/ioutil/ioutil.go
index 8aabf699e..87c628b6e 100644
--- a/internal/ioutil/ioutil.go
+++ b/internal/ioutil/ioutil.go
@@ -127,6 +127,17 @@ var deadlineWorkPanics atomic.Int64
// If the deadline/context passes before the function finishes executing,
// the zero value and the context error is returned.
func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
+ return withDeadline(ctx, timeout, work, false)
+}
+
+// WithDeadlineAndWait cancels work at the deadline, but waits for it to finish
+// and returns its actual result. Use it for mutations whose caller must not
+// release a lock or start cleanup while the work can still change storage.
+func WithDeadlineAndWait[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
+ return withDeadline(ctx, timeout, work, true)
+}
+
+func withDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error), wait bool) (result V, err error) {
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -163,6 +174,10 @@ func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(c
case v := <-c:
return v.val, v.err
case <-ctx.Done():
+ if wait {
+ v := <-c
+ return v.val, v.err
+ }
var zero V
return zero, ctx.Err()
}
diff --git a/internal/ioutil/ioutil_test.go b/internal/ioutil/ioutil_test.go
index b71c2f2e0..3e9984039 100644
--- a/internal/ioutil/ioutil_test.go
+++ b/internal/ioutil/ioutil_test.go
@@ -25,9 +25,62 @@ import (
"os"
"strings"
"testing"
+ "testing/synctest"
"time"
)
+func TestWithDeadlineAndWait(t *testing.T) {
+ for _, deadline := range []bool{false, true} {
+ name := "cancel"
+ if deadline {
+ name = "deadline"
+ }
+ t.Run(name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ entered, release := make(chan struct{}), make(chan struct{})
+ done := make(chan ioret[int], 1)
+ go func() {
+ v, err := WithDeadlineAndWait(ctx, time.Second, func(context.Context) (int, error) {
+ close(entered)
+ <-release
+ return 42, nil
+ })
+ done <- ioret[int]{v, err}
+ }()
+ <-entered
+ if deadline {
+ time.Sleep(2 * time.Second)
+ } else {
+ cancel()
+ }
+ synctest.Wait()
+ select {
+ case r := <-done:
+ t.Fatalf("returned before mutation completed: %+v", r)
+ default:
+ }
+ close(release)
+ synctest.Wait()
+ r := <-done
+ if r.err != nil || r.val != 42 {
+ t.Fatalf("lost actual commit result: %+v", r)
+ }
+ })
+ })
+ }
+}
+
+func TestWithDeadlineAndWaitPanic(t *testing.T) {
+ _, err := WithDeadlineAndWait(t.Context(), time.Second, func(context.Context) (int, error) {
+ panic("mutation failure")
+ })
+ if err == nil || !strings.Contains(err.Error(), "mutation failure") {
+ t.Fatalf("worker panic was not returned: %v", err)
+ }
+}
+
type sleepWriter struct {
timeout time.Duration
}