astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-11-10 01:31:52 +03:00
Code Issues Packages Projects Releases Wiki Activity

Integrate complete native restore repair after R2 cleanup safety

Browse Source 
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-11-03 21:20:24 +08:00
parent 18ba0ef32a fa42a09826
commit fcbf2c7d49
5 changed files with 778 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/erasure-server-pool.go
+31 -12
View File
@@ -1653,7 +1653,17 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
return oi, err
}
poolIdx, err := z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
var poolIdx int
if dstOpts.MetadataChg && cpSrcDstSame && srcInfo.metadataOnly && srcOpts.VersionID == dstOpts.VersionID {
// Restore status updates address an existing version. Keep ordinary
// copy/write allocation unchanged, including its latest-version lookup.
poolIdx, err = z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{
VersionID: dstOpts.VersionID, MetadataChg: true, NoLock: true,
SkipDecommissioned: true, SkipRebalancing: true,
})
} else {
poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
}
if err != nil {
return objInfo, err
}
@@ -2340,18 +2350,24 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
}
}()
if !z.SinglePool() {
if !opts.NoLock {
lk := z.NewNSLock(bucket, encodeDirObject(object))
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
if err != nil {
return objInfo, err
}
ctx = lkctx.Context()
defer lk.Unlock(lkctx)
// Keep object -> upload lock ordering for single-pool completion too:
// native restore holds the object lock while publishing upload parts.
if !opts.NoLock {
lk := z.NewNSLock(bucket, encodeDirObject(object))
if z.SinglePool() {
// Preserve the set-local lock domain used by single-pool writes.
lk = z.serverPools[0].getHashedSet(object).NewNSLock(bucket, encodeDirObject(object))
}
opts.NoLock = true
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
if err != nil {
return objInfo, err
}
ctx = lkctx.Context()
defer lk.Unlock(lkctx)
}
opts.NoLock = true
if !z.SinglePool() {
// A conditional completion must be evaluated against the logical
// latest object across pools, under the object write lock held for
// this operation. The pool hosting the upload may only hold a stale
@@ -3358,7 +3374,10 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck
// Avoid restoring object from a pool being decommissioned.
opts.SkipDecommissioned = true
opts.NoLock = true
idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
// Restore an existing version, not the latest version's write location.
lookupOpts := opts
lookupOpts.MetadataChg = true
idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, lookupOpts)
if err != nil {
return err
}