From cd1f6cec3c22f6c5f6b9f10d0c3669fc301f3a5d Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 25 Oct 2026 00:19:42 +0800 Subject: [PATCH] fix: drain canceled storage commits before releasing object locks Wait for admitted data and metadata mutations, preserve overwrite backups during rollback, and reject rollback of a different generation. Report conflicting or unreadable pool metadata as read-quorum failures instead of exposing stale copies. Add barrier-driven local/RPC and multi-pool regressions for cancellation, self-copy, rollback and old-value reads. Signed-off-by: Feng Ruohang --- cmd/erasure-object.go | 21 +- cmd/erasure-read-generation_test.go | 125 ++++++++ cmd/erasure-server-pool.go | 11 + cmd/rename-cancellation_test.go | 454 ++++++++++++++++++++++++++++ cmd/storage-rest-client.go | 30 +- cmd/xl-storage-disk-id-check.go | 20 +- cmd/xl-storage.go | 26 +- internal/ioutil/ioutil.go | 15 + internal/ioutil/ioutil_test.go | 53 ++++ 9 files changed, 730 insertions(+), 25 deletions(-) create mode 100644 cmd/erasure-read-generation_test.go create mode 100644 cmd/rename-cancellation_test.go 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 }