mirror of
https://github.com/pgsty/minio.git
synced 2026-11-10 17:45:32 +03:00
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 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
+14
-1
@@ -2675,6 +2675,19 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje
|
|||||||
return nil
|
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
|
// 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
|
// 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 {
|
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
|
// Single-pool and direct callers must protect the metadata snapshot
|
||||||
// through download, rehydration and failure cleanup, just like pools.
|
// through download, rehydration and failure cleanup, just like pools.
|
||||||
lk := er.NewNSLock(bucket, object)
|
lk := er.NewNSLock(bucket, object)
|
||||||
lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
|
lkctx, err := getRestoreLock(ctx, lk)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3371,7 +3371,7 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck
|
|||||||
if !opts.NoLock {
|
if !opts.NoLock {
|
||||||
// Acquire write lock before restoring transitioned object
|
// Acquire write lock before restoring transitioned object
|
||||||
lk := z.NewNSLock(bucket, object)
|
lk := z.NewNSLock(bucket, object)
|
||||||
lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
|
lkctx, err := getRestoreLock(ctx, lk)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -319,6 +319,8 @@ var (
|
|||||||
|
|
||||||
globalOperationTimeout = newDynamicTimeout(10*time.Minute, 5*time.Minute) // default timeout for general ops
|
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
|
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.
|
// Listing recovery has its own failure history and a five-second caller budget.
|
||||||
globalListingOperationTimeout = newDynamicTimeout(5*time.Second, 5*time.Second)
|
globalListingOperationTimeout = newDynamicTimeout(5*time.Second, 5*time.Second)
|
||||||
|
|
||||||
|
|||||||
@@ -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) {
|
func TestNativeRestoreConcurrentCompletion(t *testing.T) {
|
||||||
for _, separateSets := range []bool{false, true} {
|
for _, separateSets := range []bool{false, true} {
|
||||||
t.Run(fmt.Sprintf("separate-set-locks=%v", separateSets), func(t *testing.T) {
|
t.Run(fmt.Sprintf("separate-set-locks=%v", separateSets), func(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user