From 82dd12faf8f2439a2a84ed6a985e55d97cff7276 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 4 Nov 2026 16:37:42 +0800 Subject: [PATCH] fix(restore): keep queued restores waiting for the object lock The POST handler marks a version ongoing before its background restore takes the object lock, and only the restore clears that marker. Native restore holds the lock through the tier download, so a restore of another version of the same key waits behind it. Both restore entry points gave up after one delete-operation timeout and returned before any cleanup, leaving ongoing-request="true" with no owner. Every later POST for that version was refused with 409 and nothing expired the marker. Retry the lock while the restore's context lives, with a dedicated dynamic timeout so repeated waits do not lengthen delete timeouts. Shutdown or a caller deadline still ends the wait. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang --- cmd/erasure-object.go | 15 ++++++- cmd/erasure-server-pool.go | 2 +- cmd/globals.go | 2 + cmd/restore-native_test.go | 80 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 97 insertions(+), 2 deletions(-) diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index e3171aaca..64658aa50 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2675,6 +2675,19 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje return nil } +// getRestoreLock waits for a restore's object lock for as long as ctx lives. +// The POST handler has already marked the version ongoing, and only the +// restore clears that marker. A restore queued behind another version's +// download must therefore keep waiting rather than give up after one timeout. +func getRestoreLock(ctx context.Context, lk RWLocker) (LockContext, error) { + for { + lkctx, err := lk.GetLock(ctx, globalRestoreLockTimeout) + if !errors.As(err, &OperationTimedOut{}) || ctx.Err() != nil { + return lkctx, err + } + } +} + // restoreTransitionedObject for multipart object chunks the file stream from remote tier into the same number of parts // as in the xl.meta for this version and rehydrates the part.n into the fi.DataDir for this version as in the xl.meta func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket string, object string, opts ObjectOptions) error { @@ -2682,7 +2695,7 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s // Single-pool and direct callers must protect the metadata snapshot // through download, rehydration and failure cleanup, just like pools. lk := er.NewNSLock(bucket, object) - lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout) + lkctx, err := getRestoreLock(ctx, lk) if err != nil { return err } diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index e8678ad92..2f9f0d7b4 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -3371,7 +3371,7 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck if !opts.NoLock { // Acquire write lock before restoring transitioned object lk := z.NewNSLock(bucket, object) - lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout) + lkctx, err := getRestoreLock(ctx, lk) if err != nil { return err } diff --git a/cmd/globals.go b/cmd/globals.go index e71294508..607a2ecf5 100644 --- a/cmd/globals.go +++ b/cmd/globals.go @@ -319,6 +319,8 @@ var ( globalOperationTimeout = newDynamicTimeout(10*time.Minute, 5*time.Minute) // default timeout for general ops globalDeleteOperationTimeout = newDynamicTimeout(5*time.Minute, 1*time.Minute) // default time for delete ops + // Queued restores retry this wait; keep their failures out of delete timeouts. + globalRestoreLockTimeout = newDynamicTimeout(5*time.Minute, 1*time.Minute) // Listing recovery has its own failure history and a five-second caller budget. globalListingOperationTimeout = newDynamicTimeout(5*time.Second, 5*time.Second) diff --git a/cmd/restore-native_test.go b/cmd/restore-native_test.go index b01656867..884b88d24 100644 --- a/cmd/restore-native_test.go +++ b/cmd/restore-native_test.go @@ -603,6 +603,86 @@ func TestNativeRestoreSerializesObjectLock(t *testing.T) { } } +// The POST handler marks a version ongoing before its background restore +// takes the object lock, and only that restore clears the marker. A restore +// queued behind another version's download must keep waiting: one that gave up +// after a lock timeout would leave the marker for every later POST to refuse. +// The wait still ends with its context. +func TestNativeRestoreQueuedBehindDownload(t *testing.T) { + for _, pools := range []int{1, 2} { + t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) { + r := newRestoreSafetyTest(t, pools, versioning.Enabled) + first, _ := nativeRestoreSeed(t, r, 0, false, false) + queued, body := nativeRestoreSeed(t, r, pools-1, false, false) + prev := globalRestoreLockTimeout + globalRestoreLockTimeout = newDynamicTimeout(50*time.Millisecond, 50*time.Millisecond) + t.Cleanup(func() { globalRestoreLockTimeout = prev }) + tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier) + blocked := &nativeRestoreBlockedTier{WarmBackend: tier, entered: make(chan struct{}), release: make(chan struct{})} + globalTierConfigMgr.drivercache["WARM"] = blocked + var once sync.Once + unblock := func() { once.Do(func() { close(blocked.release) }) } + defer unblock() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + firstDone := make(chan error, 1) + go func() { + firstDone <- r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, first.VersionID)) + }() + select { + case <-blocked.entered: + case <-ctx.Done(): + t.Fatal("first restore never reached the tier") + } + short, shortCancel := context.WithTimeout(ctx, 200*time.Millisecond) + defer shortCancel() + ended := make(chan error, 1) + go func() { + ended <- r.z.RestoreTransitionedObject(short, r.bucket, r.object, nativeRestoreOptions(r, queued.VersionID)) + }() + select { + case err := <-ended: + if err == nil { + t.Fatal("restore finished without the object lock") + } + case <-time.After(5 * time.Second): + t.Fatal("restore kept waiting for the lock after its context ended") + } + queuedDone := make(chan error, 1) + go func() { + queuedDone <- r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, queued.VersionID)) + }() + select { + case err := <-queuedDone: + t.Fatalf("queued restore gave up behind the download: %v", err) + case <-time.After(10 * 50 * time.Millisecond): + } + unblock() + for _, done := range []chan error{firstDone, queuedDone} { + select { + case err := <-done: + if err != nil { + t.Fatalf("restore: %v", err) + } + case <-ctx.Done(): + t.Fatal("restore did not finish") + } + } + info, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(queued.VersionID)) + if err != nil || info.RestoreOngoing { + t.Fatalf("queued version still ongoing after restore: %q, %v", info.UserDefined[xhttp.AmzRestore], err) + } + reads := tier.gets.Load() + if _, got, err := r.read(queued.VersionID); err != nil || !bytes.Equal(got, body) { + t.Fatalf("queued restore contents: %v", err) + } + if tier.gets.Load() != reads { + t.Error("queued version was not restored locally") + } + }) + } +} + func TestNativeRestoreConcurrentCompletion(t *testing.T) { for _, separateSets := range []bool{false, true} { t.Run(fmt.Sprintf("separate-set-locks=%v", separateSets), func(t *testing.T) {