From 32ab506924a9500eee9fa108e79947a74e3a0fc6 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 03:31:03 +0800 Subject: [PATCH] fix(pools): lock single-pool object writes in the object's own erasure set In a single pool, DELETE in every form, bulk delete, the CopyObject destination, SelectObjectContent, batch key rotation and the upload-ID guards of complete, abort, get-info and list-parts locked through the first erasure set, while PUT, multipart commit, PutObjectPart, tag, retention, legal-hold and metadata updates, restore, transition and heal lock in the set the object hashes to. When the sets of a pool use different hosts, as every automatic layout of 18 or more hosts does, the two locks ask different lockers and do not exclude each other, and concurrent read-modify-writes of xl.meta can lose acknowledged versions and delete markers or leave an object on a minority of drives. Route these locks through NewObjectNSLock, which returns the lock of the object's own set in a single pool and the pools lock otherwise. Upload guards choose the set by their object and keep the name object/uploadID. True prefix deletion and the locks of names that are not objects keep their domain. Bulk delete groups its names by set and takes one set's lock at a time, in ascending set index; a failed group fails alone. A single node, whose sets share one local lock table, keeps its one bulk lock. The COPY handler takes its source and destination locks before it opens the source, sorted, and transactionally: bounded waits while holding a lock, release in reverse order and retry with new lock instances, one deadline. It reads the source under them, passes NoLock to every nested call, and runs that work in the locks' context. This also removes a deadlock of copies in opposite directions wherever both locks were in one domain, such as a single node or several pools. Lock names and RPCs are unchanged, but a pool is protected only when every node runs this change: affected pools must be upgraded and rolled back with a drained restart. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang --- cmd/batch-rotate.go | 2 +- cmd/erasure-server-pool-lock-domain_test.go | 1199 +++++++++++++++++++ cmd/erasure-server-pool-locks.go | 322 +++++ cmd/erasure-server-pool-locks_test.go | 555 +++++++++ cmd/erasure-server-pool.go | 62 +- cmd/metacache-server-pool.go | 9 +- cmd/object-api-interface.go | 6 + cmd/object-handlers.go | 61 +- cmd/object-retention-authz_test.go | 12 + 9 files changed, 2194 insertions(+), 34 deletions(-) create mode 100644 cmd/erasure-server-pool-lock-domain_test.go create mode 100644 cmd/erasure-server-pool-locks.go create mode 100644 cmd/erasure-server-pool-locks_test.go diff --git a/cmd/batch-rotate.go b/cmd/batch-rotate.go index a238af08b..3ebbc844d 100644 --- a/cmd/batch-rotate.go +++ b/cmd/batch-rotate.go @@ -180,7 +180,7 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob versioned := globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject) versionSuspended := globalBucketVersioningSys.PrefixSuspended(srcBucket, srcObject) - lock := api.NewNSLock(r.Bucket, objInfo.Name) + lock := api.NewObjectNSLock(r.Bucket, objInfo.Name) lkctx, err := lock.GetLock(ctx, globalOperationTimeout) if err != nil { return err diff --git a/cmd/erasure-server-pool-lock-domain_test.go b/cmd/erasure-server-pool-lock-domain_test.go new file mode 100644 index 000000000..f2a305cb3 --- /dev/null +++ b/cmd/erasure-server-pool-lock-domain_test.go @@ -0,0 +1,1199 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage stack +// +// 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" + "crypto/md5" + "encoding/base64" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/amztime" + objectlock "github.com/minio/minio/internal/bucket/object/lock" + "github.com/minio/minio/internal/bucket/versioning" + "github.com/minio/minio/internal/crypto" + "github.com/minio/minio/internal/dsync" + "github.com/minio/minio/internal/hash" + xhttp "github.com/minio/minio/internal/http" +) + +// A single pool whose two erasure sets share the test drives, but lock as the +// sets of a distributed pool do whose sets use different hosts. With one +// shared local table the sets are one lock domain, as on a single node. +const ( + ldShared = "shared" // one local lock table for both sets + ldLocal = "local-split" // a local lock table per set + ldDsync = "dsync-disjoint" // in-process dsync, no locker in common + ldOverlap = "dsync-overlap" // in-process dsync, one of three lockers in common +) + +var ldModes = []string{ldShared, ldLocal, ldDsync, ldOverlap} + +// ldWait is how long an operation must stay blocked behind a held commit. +const ldWait = 400 * time.Millisecond + +type ldGateKey struct{} + +// ldGate holds the storage commit of the operation whose context carries it. +type ldGate struct { + entered, release chan struct{} + once, open sync.Once +} + +func newLDGate() *ldGate { + return &ldGate{entered: make(chan struct{}), release: make(chan struct{})} +} + +func (g *ldGate) let() { g.open.Do(func() { close(g.release) }) } + +func ldHold(ctx context.Context) { + g, _ := ctx.Value(ldGateKey{}).(*ldGate) + if g == nil { + return + } + g.once.Do(func() { close(g.entered) }) + select { + case <-g.release: + case <-ctx.Done(): + } +} + +// ldGateDisk holds every metadata commit of a gated operation. +type ldGateDisk struct{ StorageAPI } + +func (d ldGateDisk) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (RenameDataResp, error) { + ldHold(ctx) + return d.StorageAPI.RenameData(ctx, srcVolume, srcPath, fi, dstVolume, dstPath, opts) +} + +func (d ldGateDisk) RenamePart(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string, meta []byte, skipParent string) error { + ldHold(ctx) + return d.StorageAPI.RenamePart(ctx, srcVolume, srcPath, dstVolume, dstPath, meta, skipParent) +} + +func (d ldGateDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) error { + ldHold(ctx) + return d.StorageAPI.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) +} + +func (d ldGateDisk) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) []error { + ldHold(ctx) + return d.StorageAPI.DeleteVersions(ctx, volume, versions, opts) +} + +func (d ldGateDisk) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error { + ldHold(ctx) + return d.StorageAPI.WriteMetadata(ctx, origvolume, volume, path, fi) +} + +func (d ldGateDisk) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error { + ldHold(ctx) + return d.StorageAPI.UpdateMetadata(ctx, volume, path, fi, opts) +} + +// ldSplit turns the single pool of z into two sets that share its drives and +// lock as mode says. Their drives hold gated commits. +func ldSplit(t *testing.T, z *erasureServerPools, mode string) { + t.Helper() + pool := z.serverPools[0] + sets, count := pool.sets, pool.setCount + t.Cleanup(func() { pool.sets, pool.setCount = sets, count }) + first, second := *sets[0], *sets[0] + second.setIndex = 1 + switch mode { + case ldShared: + case ldLocal: + first.nsMutex, second.nsMutex = newNSLock(false), newNSLock(false) + case ldDsync, ldOverlap: + l := make([]dsync.NetLocker, 8) + for i := range l { + l[i] = newLocker() + } + a, b := l[:4], l[4:] + if mode == ldOverlap { + // Each set can grant its quorum of two without the shared one. + a, b = l[:3], l[2:5] + } + first.nsMutex, second.nsMutex = newNSLock(true), newNSLock(true) + first.getLockers = func() ([]dsync.NetLocker, string) { return slices.Clone(a), "lock-domain-test" } + second.getLockers = func() ([]dsync.NetLocker, string) { return slices.Clone(b), "lock-domain-test" } + default: + t.Fatalf("unknown mode %q", mode) + } + disks := sets[0].getDisks + gated := func() []StorageAPI { + ds := disks() + for i, d := range ds { + if d != nil { + ds[i] = ldGateDisk{d} + } + } + return ds + } + first.getDisks, second.getDisks = gated, gated + pool.sets, pool.setCount = []*erasureObjects{&first, &second}, 2 +} + +// ldPool returns a split single pool and its unversioned test bucket. +func ldPool(t *testing.T, mode string) (*erasureServerPools, string) { + t.Helper() + z, bucket := bulkDeletePools(t, 1) + ldSplit(t, z, mode) + return z, bucket +} + +// ldKey returns a name with prefix that hashes to set. +func ldKey(t *testing.T, z *erasureServerPools, prefix string, set int) string { + t.Helper() + for i := range 1000 { + key := fmt.Sprintf("%s-%d", prefix, i) + if z.serverPools[0].getHashedSetIndex(key) == set { + return key + } + } + t.Fatalf("no name with prefix %q hashes to set %d", prefix, set) + return "" +} + +// ldOverlap runs hold until it is inside its storage commit, then runs other. +// It reports whether other was still waiting after ldWait, then lets hold +// commit and returns both results. +func ldOverlapRun(t *testing.T, hold, other func(context.Context) error) (holdErr, otherErr error, waited bool) { + t.Helper() + g := newLDGate() + defer g.let() + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Minute) + defer cancel() + held := make(chan error, 1) + go func() { held <- hold(context.WithValue(ctx, ldGateKey{}, g)) }() + select { + case <-g.entered: + case err := <-held: + t.Fatalf("held operation finished before its commit: %v", err) + case <-ctx.Done(): + t.Fatal("held operation never reached its commit") + } + done := make(chan error, 1) + go func() { done <- other(ctx) }() + select { + case otherErr = <-done: + case <-time.After(ldWait): + waited = true + } + g.let() + holdErr = <-held + if waited { + otherErr = <-done + } + return holdErr, otherErr, waited +} + +func ldReader(r io.Reader, size int64) *PutObjReader { + hr, err := hash.NewReader(context.Background(), r, size, "", "", size) + if err != nil { + panic(err) + } + return NewPutObjReader(hr) +} + +func ldPut(ctx context.Context, z *erasureServerPools, bucket, key, body string) error { + _, err := z.PutObject(ctx, bucket, key, ldReader(strings.NewReader(body), int64(len(body))), ObjectOptions{}) + return err +} + +func ldMustPut(t *testing.T, z *erasureServerPools, bucket, key, body string) { + t.Helper() + if err := ldPut(t.Context(), z, bucket, key, body); err != nil { + t.Fatal(err) + } +} + +func ldBody(t *testing.T, z *erasureServerPools, bucket, key string) (string, ObjectInfo) { + t.Helper() + gr, err := z.GetObjectNInfo(t.Context(), bucket, key, nil, nil, ObjectOptions{}) + if err != nil { + t.Fatalf("read %s: %v", key, err) + } + defer gr.Close() + body, err := io.ReadAll(gr) + if err != nil { + t.Fatal(err) + } + return string(body), gr.ObjInfo +} + +func ldAbsent(t *testing.T, z *erasureServerPools, bucket, key string) { + t.Helper() + if _, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}); !isErrObjectNotFound(err) { + t.Errorf("%s: %v, want not found", key, err) + } +} + +// ldMetadataCopy is a metadata-only self copy of key from a snapshot read +// without its lock, as the COPY handler read a copy onto itself. +func ldMetadataCopy(ctx context.Context, z *erasureServerPools, bucket, key, meta string) error { + oi, err := z.GetObjectInfo(ctx, bucket, key, ObjectOptions{NoLock: true}) + if err != nil { + return err + } + oi.metadataOnly = true + oi.UserDefined = cloneMSS(oi.UserDefined) + oi.UserDefined["x-amz-meta-copied"] = meta + _, err = z.CopyObject(ctx, bucket, key, bucket, key, oi, ObjectOptions{}, ObjectOptions{}) + return err +} + +// In one pool, a DELETE, bulk DELETE, COPY or tag update of a key must wait for +// a write of the same key that is inside its commit, and the reverse, whatever +// the hosts of the key's set are. +func TestLockDomainSinglePoolWritersExclude(t *testing.T) { + type tc struct { + name string + setup func(t *testing.T, z *erasureServerPools, bucket, key string) + hold, other func(ctx context.Context, z *erasureServerPools, bucket, key string) error + check func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) + } + seed := func(t *testing.T, z *erasureServerPools, bucket, key string) { ldMustPut(t, z, bucket, key, "seed") } + put := func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + return ldPut(ctx, z, bucket, key, "written") + } + noErrs := func(t *testing.T, holdErr, otherErr error) { + t.Helper() + if holdErr != nil || otherErr != nil { + t.Fatalf("held: %v, other: %v", holdErr, otherErr) + } + } + cases := []tc{{ + name: "put/delete", + setup: seed, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + ldAbsent(t, z, bucket, key) + }, + }, { + // The If-Match check and the delete are atomic with writers: the + // condition is checked against the PUT's version. + name: "put/delete-if-match", + setup: seed, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + seedETag := getMD5Hash([]byte("seed")) + _, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != seedETag }}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + if holdErr != nil || !isErrPreconditionFailed(otherErr) { + t.Fatalf("put: %v; If-Match delete of the replaced version: %v, want precondition failed", holdErr, otherErr) + } + if body, _ := ldBody(t, z, bucket, key); body != "written" { + t.Errorf("body %q, want the PUT's", body) + } + }, + }, { + // ILM's delete-all deletes one key through prefix deletion. + name: "put/delete-all", + setup: seed, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{DeletePrefix: true, DeletePrefixObject: true}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + ldAbsent(t, z, bucket, key) + }, + }, { + name: "put/bulk-delete", + setup: seed, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, errs := z.DeleteObjects(ctx, bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key}}}, ObjectOptions{}) + return errs[0] + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + ldAbsent(t, z, bucket, key) + }, + }, { + name: "put/copy-metadata", + setup: seed, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + return ldMetadataCopy(ctx, z, bucket, key, "1") + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + if body, _ := ldBody(t, z, bucket, key); body != "written" { + t.Errorf("body %q after copy, want the PUT's", body) + } + }, + }, { + name: "put/copy-rewrite", + setup: func(t *testing.T, z *erasureServerPools, bucket, key string) { + ldMustPut(t, z, bucket, key, "seed") + ldMustPut(t, z, bucket, key+"-src", "copied") + }, + hold: put, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + gr, err := z.GetObjectNInfo(ctx, bucket, key+"-src", nil, nil, ObjectOptions{}) + if err != nil { + return err + } + defer gr.Close() + src := gr.ObjInfo + src.PutObjReader = ldReader(gr, src.Size) + _, err = z.CopyObject(ctx, bucket, key+"-src", bucket, key, src, ObjectOptions{}, ObjectOptions{}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + if body, _ := ldBody(t, z, bucket, key); body != "copied" { + t.Errorf("body %q, want the copy's, which ran after the PUT", body) + } + }, + }, { + name: "delete/tags", + setup: seed, + hold: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{}) + return err + }, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, err := z.PutObjectTags(ctx, bucket, key, "k=v", ObjectOptions{}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + if holdErr != nil || !isErrObjectNotFound(otherErr) { + t.Fatalf("delete: %v; tags: %v, want not found", holdErr, otherErr) + } + ldAbsent(t, z, bucket, key) + }, + }, { + name: "copy/tags", + setup: seed, + hold: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + return ldMetadataCopy(ctx, z, bucket, key, "1") + }, + other: func(ctx context.Context, z *erasureServerPools, bucket, key string) error { + _, err := z.PutObjectTags(ctx, bucket, key, "k=v", ObjectOptions{}) + return err + }, + check: func(t *testing.T, z *erasureServerPools, bucket, key string, holdErr, otherErr error) { + noErrs(t, holdErr, otherErr) + oi, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if err != nil || oi.UserTags != "k=v" || oi.UserDefined["x-amz-meta-copied"] != "1" { + t.Errorf("after copy then tags: tags %q, meta %q, %v", oi.UserTags, oi.UserDefined["x-amz-meta-copied"], err) + } + }, + }} + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + key := ldKey(t, z, strings.ReplaceAll(c.name, "/", "-"), 1) + c.setup(t, z, bucket, key) + holdErr, otherErr, waited := ldOverlapRun(t, + func(ctx context.Context) error { return c.hold(ctx, z, bucket, key) }, + func(ctx context.Context) error { return c.other(ctx, z, bucket, key) }) + if !waited { + t.Errorf("%s ran while %s held its commit (other: %v)", c.name[strings.Index(c.name, "/")+1:], c.name[:strings.Index(c.name, "/")], otherErr) + } + c.check(t, z, bucket, key, holdErr, otherErr) + }) + } + }) + } +} + +// ldUpload starts an upload of key whose upload resource, and that of its +// first part, hash to the set of key when sameSet is set. +func ldUpload(t *testing.T, z *erasureServerPools, bucket, key string, sameSet bool) string { + t.Helper() + pool := z.serverPools[0] + for range 64 { + res, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + set := pool.getHashedSetIndex(key) + if !sameSet || (pool.getHashedSetIndex(pathJoin(key, res.UploadID)) == set && + pool.getHashedSetIndex(pathJoin(key, res.UploadID, "1")) == set) { + return res.UploadID + } + if err := z.AbortMultipartUpload(t.Context(), bucket, key, res.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + t.Fatal("no upload ID with the wanted hashes") + return "" +} + +// The upload-ID guards of complete and abort exclude a PutObjectPart of the +// same upload. Get-info and list-parts read, as PutObjectPart's own guard +// does, and do not wait for it. +func TestLockDomainUploadGuards(t *testing.T) { + part := []byte("the only part") + partETag := getMD5Hash(part) + putPart := func(ctx context.Context, z *erasureServerPools, bucket, key, id string) error { + _, err := z.PutObjectPart(ctx, bucket, key, id, 1, ldReader(bytes.NewReader(part), int64(len(part))), ObjectOptions{}) + return err + } + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + for _, op := range []string{"complete", "abort", "info", "list"} { + t.Run(op, func(t *testing.T) { + key := ldKey(t, z, "upload-"+op, 1) + id := ldUpload(t, z, bucket, key, false) + holdErr, otherErr, waited := ldOverlapRun(t, + func(ctx context.Context) error { return putPart(ctx, z, bucket, key, id) }, + func(ctx context.Context) error { + var err error + switch op { + case "complete": + _, err = z.CompleteMultipartUpload(ctx, bucket, key, id, []CompletePart{{PartNumber: 1, ETag: partETag}}, ObjectOptions{}) + case "abort": + err = z.AbortMultipartUpload(ctx, bucket, key, id, ObjectOptions{}) + case "info": + _, err = z.GetMultipartInfo(ctx, bucket, key, id, ObjectOptions{}) + case "list": + _, err = z.ListObjectParts(ctx, bucket, key, id, 0, 10, ObjectOptions{}) + } + return err + }) + exclusive := op == "complete" || op == "abort" + if waited != exclusive { + t.Errorf("%s waited=%v for a part upload inside its commit, want %v (err %v)", op, waited, exclusive, otherErr) + } + if holdErr != nil || otherErr != nil { + t.Fatalf("part: %v, %s: %v", holdErr, op, otherErr) + } + switch op { + case "complete": + if body, _ := ldBody(t, z, bucket, key); body != string(part) { + t.Errorf("completed object %q", body) + } + case "abort": + if _, err := z.GetMultipartInfo(t.Context(), bucket, key, id, ObjectOptions{}); !errors.As(err, &InvalidUploadID{}) { + t.Errorf("upload after abort: %v", err) + } + } + }) + } + }) + } +} + +// Bulk delete groups its names by set and must still wait for a completion +// of one of them, also when the batch names the completion's upload +// resource itself; neither may hold the other. +func TestLockDomainBulkDeleteAgainstUploads(t *testing.T) { + part := []byte("the only part") + partETag := getMD5Hash(part) + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + for _, alias := range []bool{false, true} { + t.Run(fmt.Sprintf("complete/alias=%v", alias), func(t *testing.T) { + key := ldKey(t, z, fmt.Sprintf("bulk-complete-%v", alias), 1) + id := ldUpload(t, z, bucket, key, false) + if _, err := z.PutObjectPart(t.Context(), bucket, key, id, 1, mustGetPutObjReader(t, bytes.NewReader(part), int64(len(part)), "", ""), ObjectOptions{}); err != nil { + t.Fatal(err) + } + names := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key}}} + if alias { + names = append(names, ObjectToDelete{ObjectV: ObjectV{ObjectName: pathJoin(key, id)}}) + } + holdErr, otherErr, waited := ldOverlapRun(t, + func(ctx context.Context) error { + _, err := z.CompleteMultipartUpload(ctx, bucket, key, id, []CompletePart{{PartNumber: 1, ETag: partETag}}, ObjectOptions{}) + return err + }, + func(ctx context.Context) error { + _, errs := z.DeleteObjects(ctx, bucket, names, ObjectOptions{}) + return errors.Join(errs...) + }) + if !waited { + t.Errorf("bulk delete ran while the completion held its commit (%v)", otherErr) + } + if holdErr != nil || otherErr != nil { + t.Fatalf("complete: %v, bulk: %v", holdErr, otherErr) + } + ldAbsent(t, z, bucket, key) + }) + } + // The names alias the locks of a part upload; they are other keys + // and need not wait, but neither side may block the other. + t.Run("put-part/alias", func(t *testing.T) { + key := ldKey(t, z, "bulk-part", 1) + id := ldUpload(t, z, bucket, key, true) + names := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: pathJoin(key, id)}}, {ObjectV: ObjectV{ObjectName: pathJoin(key, id, "1")}}} + holdErr, otherErr, _ := ldOverlapRun(t, + func(ctx context.Context) error { + _, err := z.PutObjectPart(ctx, bucket, key, id, 1, ldReader(bytes.NewReader(part), int64(len(part))), ObjectOptions{}) + return err + }, + func(ctx context.Context) error { + _, errs := z.DeleteObjects(ctx, bucket, names, ObjectOptions{}) + return errors.Join(errs...) + }) + if holdErr != nil || otherErr != nil { + t.Fatalf("part: %v, bulk: %v", holdErr, otherErr) + } + if _, err := z.CompleteMultipartUpload(t.Context(), bucket, key, id, []CompletePart{{PartNumber: 1, ETag: partETag}}, ObjectOptions{}); err != nil { + t.Fatalf("upload after the bulk delete: %v", err) + } + }) + }) + } +} + +// A group of a bulk delete whose lock is held by a writer of its set fails +// alone; the other set's group commits. A canceled request fails the groups it +// has not run, keeps the ones it has, and leaves no lock behind. +func TestLockDomainBulkDeleteGroups(t *testing.T) { + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + split := mode != ldShared + for _, cancelIt := range []bool{false, true} { + t.Run(fmt.Sprintf("cancel=%v", cancelIt), func(t *testing.T) { + k0 := ldKey(t, z, fmt.Sprintf("group-%v", cancelIt), 0) + k1 := ldKey(t, z, fmt.Sprintf("group-%v", cancelIt), 1) + ldMustPut(t, z, bucket, k0, "zero") + ldMustPut(t, z, bucket, k1, "one") + // A writer of k1 in the lock domain of its writes. + writer := z.serverPools[0].getHashedSet(k1).NewNSLock(bucket, k1) + wctx, err := writer.GetLock(t.Context(), globalOperationTimeout) + if err != nil { + t.Fatal(err) + } + var once sync.Once + unlock := func() { once.Do(func() { writer.Unlock(wctx) }) } + defer unlock() + + ctx, cancel := context.WithTimeout(t.Context(), 1500*time.Millisecond) + defer cancel() + names := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: k1}}, {ObjectV: ObjectV{ObjectName: k0}}} + type result struct { + objs []DeletedObject + errs []error + } + done := make(chan result, 1) + go func() { + objs, errs := z.DeleteObjects(ctx, bucket, names, ObjectOptions{}) + done <- result{objs, errs} + }() + if cancelIt { + if split { + // Cancel once the first group has committed. + deadline := time.Now().Add(5 * time.Second) + for { + if _, err := z.GetObjectInfo(t.Context(), bucket, k0, ObjectOptions{}); isErrObjectNotFound(err) { + break + } + if time.Now().After(deadline) { + t.Fatal("set 0's group did not commit") + } + time.Sleep(5 * time.Millisecond) + } + } else { + time.Sleep(200 * time.Millisecond) + } + cancel() + } + res := <-done + if res.objs[0].ObjectName != k1 || res.objs[1].ObjectName != k0 { + t.Fatalf("results out of order: %+v", res.objs) + } + if res.errs[0] == nil { + t.Errorf("k1 deleted while its writer held the lock") + } + if cancelIt && !errors.Is(res.errs[0], context.Canceled) { + t.Errorf("k1 after cancellation: %v", res.errs[0]) + } + if split != (res.errs[1] == nil) { + t.Errorf("k0 (other set): %v, want committed=%v", res.errs[1], split) + } + unlock() + if body, _ := ldBody(t, z, bucket, k1); body != "one" { + t.Errorf("k1 = %q", body) + } + if split { + ldAbsent(t, z, bucket, k0) + } + // No lock is left behind. + lk := z.serverPools[0].getHashedSet(k1).NewNSLock(bucket, k1) + lctx, err := lk.GetLock(t.Context(), newDynamicTimeout(5*time.Second, 5*time.Second)) + if err != nil { + t.Fatalf("lock left behind: %v", err) + } + lk.Unlock(lctx) + }) + } + }) + } +} + +// ldBulkScenario runs one versioned bulk delete over names of both sets and +// describes its results and the versions left, without generated version IDs. +func ldBulkScenario(t *testing.T, mode string) []string { + t.Helper() + z, _ := ldPool(t, mode) + bucket := "lock-domain-versioned" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}); err != nil { + t.Fatal(err) + } + vput := func(key, body string) ObjectInfo { + oi, err := z.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, strings.NewReader(body), int64(len(body)), "", ""), ObjectOptions{Versioned: true}) + if err != nil { + t.Fatal(err) + } + return oi + } + a, b, c := ldKey(t, z, "map-a", 1), ldKey(t, z, "map-b", 0), ldKey(t, z, "map-c", 1) + dir := ldKey(t, z, "map-dir", 0) + "/" + va1, va2 := vput(a, "a1"), vput(a, "a2") + vb := vput(b, "b1") + vput(c, "c1") + if _, err := z.PutObject(t.Context(), bucket, dir, mustGetPutObjReader(t, strings.NewReader(""), 0, "", ""), ObjectOptions{Versioned: true}); err != nil { + t.Fatal(err) + } + absent := mustGetUUID() + invalid := strings.Repeat("x", 1100) + names := []ObjectToDelete{ + {ObjectV: ObjectV{ObjectName: a}}, // marker + {ObjectV: ObjectV{ObjectName: b, VersionID: vb.VersionID}}, // purge + {ObjectV: ObjectV{ObjectName: invalid}}, // invalid + {ObjectV: ObjectV{ObjectName: a, VersionID: va1.VersionID}}, // purge, same key + {ObjectV: ObjectV{ObjectName: c}}, // marker + {ObjectV: ObjectV{ObjectName: dir}}, // directory marker + {ObjectV: ObjectV{ObjectName: "map-missing"}}, // absent + {ObjectV: ObjectV{ObjectName: a, VersionID: va1.VersionID}}, // duplicate + {ObjectV: ObjectV{ObjectName: b, VersionID: absent}}, // absent version + } + // Names hash with the deployment ID: compare them by role. + label := strings.NewReplacer(dir, "DIR/", a, "A", b, "B", c, "C", invalid, "INVALID", + va1.VersionID, "a1", va2.VersionID, "a2", vb.VersionID, "b1", absent, "absent") + objs, errs := z.DeleteObjects(t.Context(), bucket, names, ObjectOptions{Versioned: true}) + var out []string + for i := range names { + o := objs[i] + vid := label.Replace(o.VersionID) + if len(vid) > 10 { + vid = "generated" + } + out = append(out, label.Replace(fmt.Sprintf("%d %s vid=%q marker=%v markerVid=%v found=%v err=%v", + i, o.ObjectName, vid, o.DeleteMarker, o.DeleteMarkerVersionID != "", o.found, errs[i]))) + } + for _, key := range []string{a, b, c, dir} { + info, err := z.ListObjectVersions(t.Context(), bucket, key, "", "", "", 100) + if err != nil { + t.Fatal(err) + } + var vs []string + for _, v := range info.Objects { + if v.Name != key { + continue + } + id := "other" + switch v.VersionID { + case va2.VersionID: + id = "a2" + case va1.VersionID: + id = "a1" + case vb.VersionID: + id = "b1" + } + vs = append(vs, fmt.Sprintf("%s:marker=%v:latest=%v", id, v.DeleteMarker, v.IsLatest)) + } + out = append(out, label.Replace(key)+" "+strings.Join(vs, ",")) + } + return out +} + +// Grouping by set keeps every result where its name was and gives the same +// results and the same versions as the one bulk lock of a single node. +func TestLockDomainBulkDeleteMapping(t *testing.T) { + want := ldBulkScenario(t, ldShared) + for _, mode := range ldModes[1:] { + t.Run(mode, func(t *testing.T) { + got := ldBulkScenario(t, mode) + if !slices.Equal(got, want) { + t.Errorf("results differ from the single bulk lock:\n got %q\nwant %q", got, want) + } + }) + } +} + +// A restore holds the object lock of its set for its download. DELETE and COPY +// of the object must wait for it. +func TestLockDomainRestoreExcludesDeleteAndCopy(t *testing.T) { + for _, mode := range ldModes { + for _, op := range []string{"delete", "copy"} { + t.Run(mode+"/"+op, func(t *testing.T) { + r := newRestoreSafetyTest(t, 1, versioning.Enabled) + ldSplit(t, r.z, mode) + r.object = ldKey(t, r.z, "restore-"+op, 1) + oi, _ := nativeRestoreSeed(t, r, 0, false, false) + blocked := &nativeRestoreBlockedTier{WarmBackend: globalTierConfigMgr.drivercache["WARM"], 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(), time.Minute) + defer cancel() + restored := make(chan error, 1) + go func() { + restored <- r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID)) + }() + select { + case <-blocked.entered: + case <-ctx.Done(): + t.Fatal("restore never started its download") + } + done := make(chan error, 1) + go func() { + switch op { + case "delete": + _, err := r.z.DeleteObject(ctx, r.bucket, r.object, r.opts("")) + done <- err + case "copy": + ropts := r.opts(oi.VersionID) + ropts.NoLock = true + cur, err := r.z.GetObjectInfo(ctx, r.bucket, r.object, ropts) + if err != nil { + done <- err + return + } + cur.metadataOnly = true + cur.UserDefined = cloneMSS(cur.UserDefined) + cur.UserDefined["x-amz-meta-copied"] = "1" + vopts := r.opts(oi.VersionID) + _, err = r.z.CopyObject(ctx, r.bucket, r.object, r.bucket, r.object, cur, vopts, vopts) + done <- err + } + }() + var opErr error + select { + case opErr = <-done: + t.Errorf("%s ran during the restore (%v)", op, opErr) + case <-time.After(ldWait): + unblock() + opErr = <-done + } + unblock() + if err := <-restored; err != nil { + t.Fatalf("restore: %v", err) + } + if opErr != nil { + t.Fatalf("%s after the restore: %v", op, opErr) + } + }) + } + } +} + +// ldAPI is a split single pool served by the S3 API router. +func ldAPI(t *testing.T, mode string) (*erasureServerPools, string, http.Handler) { + t.Helper() + z, bucket := ldPool(t, mode) + if err := newTestConfig(globalMinioDefaultRegion, z); err != nil { + t.Fatal(err) + } + return z, bucket, initTestAPIEndPoints(z, nil) +} + +func ldRequest(ctx context.Context, t *testing.T, router http.Handler, method, path, body string, headers map[string]string) *httptest.ResponseRecorder { + t.Helper() + req, err := newTestSignedRequestV4(method, path, int64(len(body)), strings.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req.WithContext(ctx)) + return rec +} + +func ldCopyRequest(ctx context.Context, t *testing.T, router http.Handler, bucket, src, dst string, headers map[string]string) *httptest.ResponseRecorder { + t.Helper() + h := map[string]string{xhttp.AmzCopySource: "/" + bucket + "/" + src} + for k, v := range headers { + h[k] = v + } + return ldRequest(ctx, t, router, http.MethodPut, "/"+bucket+"/"+dst, "", h) +} + +// The COPY handler locks its destination, and a copy onto itself reads its +// source, in the lock domain of the object's writes. A copy onto itself must +// read the version that the PUT before it committed. +func TestLockDomainCopyHandler(t *testing.T) { + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket, router := ldAPI(t, mode) + for _, self := range []bool{false, true} { + t.Run(fmt.Sprintf("self=%v", self), func(t *testing.T) { + key := ldKey(t, z, fmt.Sprintf("copy-handler-%v", self), 1) + src := key + "-src" + ldMustPut(t, z, bucket, key, "seed") + ldMustPut(t, z, bucket, src, "source") + var rec *httptest.ResponseRecorder + holdErr, _, waited := ldOverlapRun(t, + func(ctx context.Context) error { return ldPut(ctx, z, bucket, key, "written") }, + func(ctx context.Context) error { + if self { + rec = ldCopyRequest(ctx, t, router, bucket, key, key, map[string]string{ + xhttp.AmzMetadataDirective: "REPLACE", "x-amz-meta-copied": "1", + }) + } else { + rec = ldCopyRequest(ctx, t, router, bucket, src, key, nil) + } + return nil + }) + if !waited { + t.Errorf("COPY ran while a PUT of its destination held its commit") + } + if holdErr != nil || rec.Code != http.StatusOK { + t.Fatalf("put: %v; copy: %d %s", holdErr, rec.Code, rec.Body.String()) + } + body, oi := ldBody(t, z, bucket, key) + want := "source" + if self { + want = "written" + if oi.UserDefined["X-Amz-Meta-Copied"] != "1" { + t.Errorf("copied metadata missing: %v", oi.UserDefined) + } + } + if body != want || oi.ETag != getMD5Hash([]byte(want)) { + t.Errorf("after the copy: body %q etag %q, want %q with its own ETag", body, oi.ETag, want) + } + }) + } + }) + } +} + +// Copies in opposite directions always make progress: between keys of +// different sets, and where both locks share one domain, as on a single node +// or with several pools. +func TestLockDomainCopyHandlerReciprocal(t *testing.T) { + for _, mode := range append(slices.Clone(ldModes), "pools") { + t.Run(mode, func(t *testing.T) { + var z *erasureServerPools + var bucket string + var router http.Handler + keys := [2]string{"reciprocal-a", "reciprocal-b"} + if mode == "pools" { + z, bucket = bulkDeletePools(t, 2) + if err := newTestConfig(globalMinioDefaultRegion, z); err != nil { + t.Fatal(err) + } + router = initTestAPIEndPoints(z, nil) + } else { + z, bucket, router = ldAPI(t, mode) + keys = [2]string{ldKey(t, z, "reciprocal", 0), ldKey(t, z, "reciprocal", 1)} + } + // Not inlined: the source is read while the destination is written. + body := strings.Repeat("reciprocal copy source ", 1<<14) + for _, k := range keys { + ldMustPut(t, z, bucket, k, body) + } + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + for round := range 8 { + var wg sync.WaitGroup + codes := make([]int, 2) + start := make(chan struct{}) + for i := range keys { + wg.Add(1) + go func() { + defer wg.Done() + <-start + codes[i] = ldCopyRequest(ctx, t, router, bucket, keys[i], keys[1-i], nil).Code + }() + } + close(start) + wg.Wait() + if codes[0] != http.StatusOK || codes[1] != http.StatusOK { + t.Fatalf("round %d: reciprocal copies returned %v", round, codes) + } + } + }) + } +} + +// The COPY handler looks up its destination while it holds the destination's +// lock: Object Lock's retention check and the tier sweeper's lookup must not +// take that lock again. +func TestLockDomainCopyHandlerDestinationLookups(t *testing.T) { + prevTier := globalTierConfigMgr + t.Cleanup(func() { globalTierConfigMgr = prevTier }) + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, _, router := ldAPI(t, mode) + globalTierConfigMgr = NewTierConfigMgr() + globalTierConfigMgr.Tiers["WARM"] = madmin.TierConfig{Version: madmin.TierConfigVer, Type: madmin.S3, Name: "WARM", S3: &madmin.TierS3{Bucket: "warm"}} + bucket := "lock-domain-worm" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{LockEnabled: true, VersioningEnabled: true}); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + src := ldKey(t, z, "lookup-src", 0) + if _, err := z.PutObject(t.Context(), bucket, src, mustGetPutObjReader(t, strings.NewReader(""), 0, "", ""), ObjectOptions{Versioned: true}); err != nil { + t.Fatal(err) + } + for _, dst := range []string{ldKey(t, z, "lookup-dst", 1), ldKey(t, z, "lookup-dir", 1) + "/"} { + if rec := ldCopyRequest(ctx, t, router, bucket, src, dst, nil); rec.Code != http.StatusOK { + t.Fatalf("copy to %s: %d %s", dst, rec.Code, rec.Body.String()) + } + // Self copy, metadata only. + if rec := ldCopyRequest(ctx, t, router, bucket, dst, dst, map[string]string{ + xhttp.AmzMetadataDirective: "REPLACE", "x-amz-meta-copied": "1", + }); rec.Code != http.StatusOK { + t.Fatalf("self copy of %s: %d %s", dst, rec.Code, rec.Body.String()) + } + } + }) + } +} + +// SelectObjectContent reads under the lock of the object's writes. +func TestLockDomainSelectWaitsForWriter(t *testing.T) { + const request = `SELECT * FROM S3ObjectSQL` + + `NONE` + + `` + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket, router := ldAPI(t, mode) + key := ldKey(t, z, "select", 1) + ldMustPut(t, z, bucket, key, "seed,1\n") + var rec *httptest.ResponseRecorder + holdErr, _, waited := ldOverlapRun(t, + func(ctx context.Context) error { return ldPut(ctx, z, bucket, key, "written,2\n") }, + func(ctx context.Context) error { + rec = ldRequest(ctx, t, router, http.MethodPost, "/"+bucket+"/"+key+"?select&select-type=2", request, nil) + return nil + }) + if !waited { + t.Errorf("SELECT ran while a PUT held its commit") + } + if holdErr != nil || rec.Code != http.StatusOK || !bytes.Contains(rec.Body.Bytes(), []byte("written,2")) { + t.Fatalf("put: %v; select: %d %q", holdErr, rec.Code, rec.Body.String()) + } + }) + } +} + +// Batch key rotation locks the object in the lock domain of its writes. +func TestLockDomainKeyRotationWaitsForWriter(t *testing.T) { + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + key := ldKey(t, z, "rotate", 1) + ldMustPut(t, z, bucket, key, "seed") + job := &BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}} + // The rotation's listing found an SSE-S3 object; it reads it again + // under the lock. + listed := ObjectInfo{Bucket: bucket, Name: key, UserDefined: map[string]string{crypto.MetaSealedKeyS3: "listed"}} + _, _, waited := ldOverlapRun(t, + func(ctx context.Context) error { return ldPut(ctx, z, bucket, key, "written") }, + func(ctx context.Context) error { return job.KeyRotate(ctx, z, listed) }) + if !waited { + t.Errorf("key rotation ran while a PUT held its commit") + } + }) + } +} + +// ldCopyGate holds a COPY of one destination inside the object layer's +// CopyObject, after the handler's Object Lock check, and reports when a +// metadata update of that destination reaches the object layer. +type ldCopyGate struct { + ObjectLayer + bucket, object string + copying, updating chan struct{} + release chan struct{} + copyOnce, updatingOnce sync.Once +} + +func (g *ldCopyGate) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, info ObjectInfo, srcOpts, dstOpts ObjectOptions) (ObjectInfo, error) { + if dstBucket == g.bucket && dstObject == g.object { + g.copyOnce.Do(func() { close(g.copying) }) + select { + case <-g.release: + case <-ctx.Done(): + return ObjectInfo{}, ctx.Err() + } + } + return g.ObjectLayer.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, info, srcOpts, dstOpts) +} + +func (g *ldCopyGate) PutObjectMetadata(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + if bucket == g.bucket && object == g.object { + g.updatingOnce.Do(func() { close(g.updating) }) + } + return g.ObjectLayer.PutObjectMetadata(ctx, bucket, object, opts) +} + +// A retention or legal-hold update of a COPY destination cannot land between +// the handler's Object Lock check and the copy: it waits for the copy, which +// then succeeds, and is applied after it. +func TestLockDomainCopyExcludesObjectLockUpdates(t *testing.T) { + for _, mode := range append(slices.Clone(ldModes), "pools") { + t.Run(mode, func(t *testing.T) { + var z *erasureServerPools + if mode == "pools" { + z, _ = bulkDeletePools(t, 2) + } else { + z, _ = ldPool(t, mode) + } + if err := newTestConfig(globalMinioDefaultRegion, z); err != nil { + t.Fatal(err) + } + bucket := "lock-domain-worm-copy" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{LockEnabled: true, VersioningEnabled: true}); err != nil { + t.Fatal(err) + } + plain := initTestAPIEndPoints(z, nil) + put := func(key string) { + t.Helper() + if rec := ldRequest(t.Context(), t, plain, http.MethodPut, "/"+bucket+"/"+key, "", nil); rec.Code != http.StatusOK { + t.Fatalf("PUT %s: %d %s", key, rec.Code, rec.Body.String()) + } + } + put("copy-source/") + for _, self := range []bool{false, true} { + for _, state := range []string{"GOVERNANCE", "COMPLIANCE", "LEGAL-HOLD"} { + t.Run(fmt.Sprintf("self=%v/%s", self, state), func(t *testing.T) { + // Directory objects: Object Lock checks their null version. + key := fmt.Sprintf("copy-destination-%v-%s/", self, state) + if mode != "pools" { + for z.serverPools[0].getHashedSetIndex(encodeDirObject(key)) != 1 { + key = "x" + key + } + } + put(key) + g := &ldCopyGate{ObjectLayer: z, bucket: bucket, object: key, + copying: make(chan struct{}), updating: make(chan struct{}), release: make(chan struct{})} + setObjectLayer(g) + defer setObjectLayer(z) + router := initTestAPIEndPoints(g, nil) + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + release := sync.OnceFunc(func() { close(g.release) }) + var wg sync.WaitGroup + defer func() { release(); cancel(); wg.Wait() }() + + src := "copy-source/" + if self { + src = key + } + copied, updated := make(chan *httptest.ResponseRecorder, 1), make(chan *httptest.ResponseRecorder, 1) + wg.Add(1) + go func() { + defer wg.Done() + copied <- ldCopyRequest(ctx, t, router, bucket, src, key, map[string]string{ + xhttp.AmzMetadataDirective: "REPLACE", "x-amz-meta-copied": "1", + }) + }() + select { + case <-g.copying: + case rec := <-copied: + t.Fatalf("COPY returned before its write: %d %s", rec.Code, rec.Body.String()) + case <-ctx.Done(): + t.Fatal("COPY never reached its write") + } + + query := url.Values{"versionId": {nullVersionID}} + body := `ON` + if state == "LEGAL-HOLD" { + query.Set("legal-hold", "") + } else { + query.Set("retention", "") + body = fmt.Sprintf(`%s%s`, + state, amztime.ISO8601Format(UTCNow().Add(365*24*time.Hour))) + } + sum := md5.Sum([]byte(body)) + wg.Add(1) + go func() { + defer wg.Done() + updated <- ldRequest(ctx, t, router, http.MethodPut, makeTestTargetURL("", bucket, key, query), body, + map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])}) + }() + select { + case <-g.updating: + case rec := <-updated: + t.Fatalf("update returned before reaching the object layer: %d %s", rec.Code, rec.Body.String()) + case <-ctx.Done(): + t.Fatal("update never reached the object layer") + } + select { + case rec := <-updated: + t.Errorf("%s update ran between the COPY's Object Lock check and its write: %d", state, rec.Code) + updated <- rec + case <-time.After(ldWait): + } + release() + for name, result := range map[string]chan *httptest.ResponseRecorder{"COPY": copied, "update": updated} { + select { + case rec := <-result: + if rec.Code != http.StatusOK { + t.Errorf("%s: %d %s", name, rec.Code, rec.Body.String()) + } + case <-ctx.Done(): + t.Fatalf("%s did not finish", name) + } + } + after, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID}) + if err != nil { + t.Fatal(err) + } + if after.UserDefined["X-Amz-Meta-Copied"] != "1" { + t.Errorf("copied metadata missing: %v", after.UserDefined) + } + if state == "LEGAL-HOLD" { + if objectlock.GetObjectLegalHoldMeta(after.UserDefined).Status != objectlock.LegalHoldOn { + t.Error("legal hold applied before the copy was lost") + } + } else if string(objectlock.GetObjectRetentionMeta(after.UserDefined).Mode) != state { + t.Error("retention applied before the copy was lost") + } + }) + } + } + }) + } +} diff --git a/cmd/erasure-server-pool-locks.go b/cmd/erasure-server-pool-locks.go new file mode 100644 index 000000000..b577bf566 --- /dev/null +++ b/cmd/erasure-server-pool-locks.go @@ -0,0 +1,322 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage stack +// +// 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 ( + "context" + "errors" + "math/rand/v2" + "slices" + "strings" + "time" + + "github.com/minio/minio-go/v7/pkg/set" +) + +// Object lock domains. +// +// Every erasure set has its own lockers, one per host among its drives, and a +// distributed lock is granted by a majority of them. Two locks of one name +// exclude each other only when they ask the same lockers. In a single pool the +// writers (PutObject, multipart commit, PutObjectPart, tags, retention, legal +// hold, metadata, restore, transition, heal) lock in the set an object hashes +// to. The pools layer locks an object, and its upload IDs, in the same set, so +// that a pool whose sets use different hosts still has one lock per object. +// With several pools every object write takes the pools lock instead, which +// these helpers then return unchanged. +// +// Generic NewNSLock keeps one stable domain for names that are not objects: +// bucket and IAM metadata locks, format and leader coordination, and the +// /[replicate]/ serializer. + +// NewObjectNSLock returns the lock of one object in the lock domain that its +// writes use: in a single pool the object's own erasure set, otherwise the +// pools lock. Directory objects are named as storage names them. +func (z *erasureServerPools) NewObjectNSLock(bucket, object string) RWLocker { + if z.SinglePool() { + object = encodeDirObject(object) + } + return z.objectNSLock(bucket, object, object) +} + +// uploadIDNSLock returns the guard of one upload ID. It is chosen by the +// upload's object, as PutObjectPart locks it in that object's set; the name +// locked stays object/uploadID. +func (z *erasureServerPools) uploadIDNSLock(bucket, object, uploadID string) RWLocker { + return z.objectNSLock(bucket, object, pathJoin(object, uploadID)) +} + +// objectNSLock returns the lock of resource, a name owned by object, in the +// lock domain of object. +func (z *erasureServerPools) objectNSLock(bucket, object, resource string) RWLocker { + if z.SinglePool() { + return z.serverPools[0].getHashedSet(encodeDirObject(object)).NewNSLock(bucket, resource) + } + return z.NewNSLock(bucket, resource) +} + +// objectLockRequest is one object lock that lockObjects takes. +type objectLockRequest struct { + bucket, object string + write bool +} + +// lockObjects takes the locks of several objects, such as the source and +// destination of a copy, before any of them is read. A name that appears twice +// gets one lock, a write lock if either asks for one. The locks are taken in +// one order, by lock domain and then by bucket and object, and transactionally +// (see getNSLocks), so that two copies in opposite directions, or an operation +// that orders its locks differently, cannot hold each other forever. The +// returned context ends when any of the locks is lost; the returned function +// releases them all. +func (z *erasureServerPools) lockObjects(ctx context.Context, reqs ...objectLockRequest) (LockContext, func(), error) { + type entry struct { + objectLockRequest + domain int + } + locks := make([]entry, 0, len(reqs)) + for _, r := range reqs { + r.object = encodeDirObject(r.object) + i := slices.IndexFunc(locks, func(e entry) bool { return e.bucket == r.bucket && e.object == r.object }) + if i >= 0 { + locks[i].write = locks[i].write || r.write + continue + } + domain := 0 + if z.SinglePool() { + domain = z.serverPools[0].getHashedSetIndex(r.object) + } + locks = append(locks, entry{objectLockRequest: r, domain: domain}) + } + slices.SortFunc(locks, func(a, b entry) int { + if a.domain != b.domain { + return a.domain - b.domain + } + return strings.Compare(pathJoin(a.bucket, a.object), pathJoin(b.bucket, b.object)) + }) + nsLocks := make([]nsLockRequest, len(locks)) + for i, l := range locks { + nsLocks[i] = nsLockRequest{ + newLock: func() RWLocker { return z.NewObjectNSLock(l.bucket, l.object) }, + write: l.write, + atomic: true, + } + } + return getNSLocks(ctx, nsLocks, globalOperationTimeout) +} + +// nsLockRequest is one lock of a transactional acquisition. +type nsLockRequest struct { + // newLock returns a new instance of the lock for each attempt. An + // instance names its holder, and a distributed unlock retries in the + // background; a later attempt with the same instance could be released + // by such a retry of the attempt before. + newLock func() RWLocker + write bool + // atomic: the lock is granted as a whole or not at all. A distributed + // lock is, also over several names. The local backend takes several + // names one at a time, holding the first while it waits for the next; + // its lock context is the context it was given. + atomic bool +} + +const ( + // nsLockTryTimeout bounds each wait for a lock while holding another. + nsLockTryTimeout = 2 * time.Second + // nsLockRetryWait is the longest random pause before a new attempt. + nsLockRetryWait = 250 * time.Millisecond +) + +// getNSLocks acquires every lock of locks, in their order, or none. While it +// holds a lock it waits for the next one only up to nsLockTryTimeout; then it +// releases what it holds in reverse order, pauses, and tries again with a +// fresh context, until the timeout of all attempts. A wait that holds nothing +// can last for the rest of the timeout. Each lock is taken in the context of +// the one before, so the returned context ends when any lease is lost. +func getNSLocks(ctx context.Context, locks []nsLockRequest, timeout *dynamicTimeout) (LockContext, func(), error) { + type held struct { + lock RWLocker + write bool + lkctx LockContext + } + release := func(hs []held) { + for i := len(hs) - 1; i >= 0; i-- { + if hs[i].write { + hs[i].lock.Unlock(hs[i].lkctx) + } else { + hs[i].lock.RUnlock(hs[i].lkctx) + } + } + } + // Fail as a single lock does: a canceled request with its own error, + // anything else as a timeout. + failed := func() error { + if err := ctx.Err(); errors.Is(err, context.Canceled) { + return err + } + return OperationTimedOut{} + } + start := time.Now() + deadline := start.Add(timeout.Timeout()) + for { + hs := make([]held, 0, len(locks)) + lctx := ctx + var err error + for _, l := range locks { + wait := time.Until(deadline) + if len(hs) > 0 || !l.atomic { + wait = min(wait, nsLockTryTimeout) + } + if wait <= 0 { + err = OperationTimedOut{} + break + } + lk := l.newLock() + try := newDynamicTimeout(wait, wait) + // The local backend waits up to the timeout for each name: bound + // the whole attempt by its context instead. + tctx, cancel := lctx, context.CancelFunc(func() {}) + if !l.atomic { + tctx, cancel = context.WithTimeout(lctx, wait) + } + var lkctx LockContext + if l.write { + lkctx, err = lk.GetLock(tctx, try) + } else { + lkctx, err = lk.GetRLock(tctx, try) + } + cancel() + if err != nil { + break + } + if !l.atomic { + lkctx.ctx = lctx + } + hs = append(hs, held{lock: lk, write: l.write, lkctx: lkctx}) + lctx = lkctx.Context() + } + if err == nil { + timeout.LogSuccess(time.Since(start)) + if len(hs) == 0 { + return LockContext{ctx: ctx, cancel: func() {}}, func() {}, nil + } + return LockContext{ctx: lctx, cancel: func() {}}, func() { release(hs) }, nil + } + release(hs) + // A lost lease cancels the context of a later wait; only the + // request's own context ends the attempts. + if ctx.Err() != nil || (!errors.As(err, &OperationTimedOut{}) && !errors.Is(err, context.Canceled)) { + return LockContext{}, nil, err + } + pause := rand.N(nsLockRetryWait) + time.Millisecond + if time.Until(deadline) <= pause { + timeout.LogFailure() + return LockContext{}, nil, OperationTimedOut{} + } + t := time.NewTimer(pause) + select { + case <-ctx.Done(): + t.Stop() + return LockContext{}, nil, failed() + case <-t.C: + } + } +} + +// sharesLocalNSLock reports whether every set of the pool uses one local lock +// table, as a single node does: there all sets are one lock domain. +func (s *erasureSets) sharesLocalNSLock() bool { + first := s.sets[0].nsMutex + if first.isDistErasure { + return false + } + for _, set := range s.sets[1:] { + if set.nsMutex != first { + return false + } + } + return true +} + +// deleteObjectsInSets is the bulk delete of a single pool whose sets do not +// share one lock table. The names are grouped by the set they hash to, the +// lock domain of their writers. The groups run one after another in ascending +// set index; each takes its set's lock over all of its names in one +// acquisition, deletes its sub-batch, and releases the lock before the next +// group starts. No two sets' locks are held at once. +// +// Each group's lock is taken in the request's context. A group whose lock +// cannot be taken fails as a whole and the next group still runs; once the +// request is canceled the remaining groups fail without being attempted. The +// results keep their positions in objects. Replication, events and auditing +// stay with the callers, on the merged results. +func (z *erasureServerPools) deleteObjectsInSets(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions, dobjects []DeletedObject, derrs []error, timeout *dynamicTimeout) { + pool := z.serverPools[0] + groups := make(map[int][]int) + for i, obj := range objects { + idx := pool.getHashedSetIndex(obj.ObjectName) + groups[idx] = append(groups[idx], i) + } + order := make([]int, 0, len(groups)) + for idx := range groups { + order = append(order, idx) + } + slices.Sort(order) + + for _, idx := range order { + members := groups[idx] + fail := func(err error) { + for _, i := range members { + derrs[i] = err + } + } + if err := ctx.Err(); err != nil { + fail(err) + continue + } + er := pool.sets[idx] + names := set.NewStringSet() + batch := make([]ObjectToDelete, len(members)) + for j, i := range members { + names.Add(objects[i].ObjectName) + batch[j] = objects[i] + } + resources := names.ToSlice() + lkctx, unlock, err := getNSLocks(ctx, []nsLockRequest{{ + newLock: func() RWLocker { return er.NewNSLock(bucket, resources...) }, + write: true, + atomic: er.nsMutex.isDistErasure || len(resources) == 1, + }}, timeout) + if err != nil { + fail(err) + continue + } + deleted, errs := er.DeleteObjects(lkctx.Context(), bucket, batch, opts) + unlock() + // Merge as the pools loop of DeleteObjects does for one pool. + for j, i := range members { + if errs[j] == nil && deleted[j].found { + dobjects[i], derrs[i] = deleted[j], nil + continue + } + if derrs[i] == nil { + dobjects[i], derrs[i] = deleted[j], errs[j] + } + } + } +} diff --git a/cmd/erasure-server-pool-locks_test.go b/cmd/erasure-server-pool-locks_test.go new file mode 100644 index 000000000..696c22771 --- /dev/null +++ b/cmd/erasure-server-pool-locks_test.go @@ -0,0 +1,555 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage stack +// +// 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 ( + "context" + "errors" + "fmt" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/dsync" +) + +// ldTry reports whether lk can be taken now, and releases it again. +func ldTry(t *testing.T, lk RWLocker, write bool) bool { + t.Helper() + dt := newDynamicTimeout(100*time.Millisecond, 100*time.Millisecond) + if write { + lc, err := lk.GetLock(t.Context(), dt) + if err == nil { + lk.Unlock(lc) + } + return err == nil + } + lc, err := lk.GetRLock(t.Context(), dt) + if err == nil { + lk.RUnlock(lc) + } + return err == nil +} + +// ldFree reports whether lk can be taken within a few seconds: a distributed +// unlock completes in the background. +func ldFree(t *testing.T, lk RWLocker, write bool) bool { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !ldTry(t, lk, write) { + if time.Now().After(deadline) { + return false + } + } + return true +} + +// ldTaken waits until someone else holds lk. +func ldTaken(t *testing.T, lk RWLocker) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for ldTry(t, lk, false) { + if time.Now().After(deadline) { + t.Fatal("lock never taken") + } + time.Sleep(time.Millisecond) + } +} + +func ldHoldLock(t *testing.T, lk RWLocker) func() { + t.Helper() + lc, err := lk.GetLock(t.Context(), globalOperationTimeout) + if err != nil { + t.Fatal(err) + } + done := false + return func() { + if !done { + done = true + lk.Unlock(lc) + } + } +} + +// Object locks, and upload guards by their object, are in the set the +// object's writes lock in; with several pools they stay the pools lock. +func TestObjectNSLockRouting(t *testing.T) { + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + pool := z.serverPools[0] + key := ldKey(t, z, "route", 1) + dir := ldKey(t, z, "route-dir", 0) + "/" + for _, name := range []string{key, dir} { + release := ldHoldLock(t, z.NewObjectNSLock(bucket, name)) + encoded := encodeDirObject(name) + if ldTry(t, pool.getHashedSet(encoded).NewNSLock(bucket, encoded), false) { + t.Errorf("%s: a writer of the object's set ignores the object lock", name) + } + release() + } + id := mustGetUUID() + release := ldHoldLock(t, z.uploadIDNSLock(bucket, key, id)) + if ldTry(t, pool.getHashedSet(key).NewNSLock(bucket, pathJoin(key, id)), false) { + t.Error("PutObjectPart's guard ignores the upload guard") + } + release() + if mode != ldShared && pool.sharesLocalNSLock() { + t.Error("split sets reported as one local table") + } + if mode == ldShared && !pool.sharesLocalNSLock() { + t.Error("one local table not recognized") + } + }) + } + t.Run("pools", func(t *testing.T) { + z, bucket := bulkDeletePools(t, 2) + release := ldHoldLock(t, z.NewNSLock(bucket, "pooled")) + defer release() + if ldTry(t, z.NewObjectNSLock(bucket, "pooled"), false) || ldTry(t, z.uploadIDNSLock(bucket, "pooled", ""), false) { + t.Error("multi-pool object locks left the pools lock") + } + }) +} + +// lockObjects merges a name asked twice into one write lock and takes source +// and destination in their own sets. +func TestLockObjects(t *testing.T) { + for _, mode := range ldModes { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + pool := z.serverPools[0] + src, dst := ldKey(t, z, "lock-src", 1), ldKey(t, z, "lock-dst", 0) + lkctx, unlock, err := z.lockObjects(t.Context(), objectLockRequest{bucket: bucket, object: src}, objectLockRequest{bucket: bucket, object: dst, write: true}) + if err != nil { + t.Fatal(err) + } + if lkctx.Context().Err() != nil { + t.Fatal("lock context ended") + } + if !ldTry(t, pool.getHashedSet(src).NewNSLock(bucket, src), false) { + t.Error("source not shared with readers") + } + if ldTry(t, pool.getHashedSet(src).NewNSLock(bucket, src), true) { + t.Error("source not locked against writers") + } + if ldTry(t, pool.getHashedSet(dst).NewNSLock(bucket, dst), false) { + t.Error("destination not locked") + } + unlock() + if !ldFree(t, pool.getHashedSet(dst).NewNSLock(bucket, dst), true) || !ldFree(t, pool.getHashedSet(src).NewNSLock(bucket, src), true) { + t.Error("locks left behind") + } + + // A copy onto itself takes one write lock; a directory object is + // locked by its stored name. + self := ldKey(t, z, "lock-self", 1) + "/" + _, unlock, err = z.lockObjects(t.Context(), objectLockRequest{bucket: bucket, object: self}, objectLockRequest{bucket: bucket, object: self, write: true}) + if err != nil { + t.Fatal(err) + } + encoded := encodeDirObject(self) + if ldTry(t, pool.getHashedSet(encoded).NewNSLock(bucket, encoded), false) { + t.Error("self copy not write locked") + } + unlock() + }) + } +} + +// ldLeaseLock is a lock whose lease the test ends. +type ldLeaseLock struct { + RWLocker + lost chan context.CancelFunc +} + +func (l ldLeaseLock) GetLock(ctx context.Context, timeout *dynamicTimeout) (LockContext, error) { + lc, err := l.RWLocker.GetLock(ctx, timeout) + if err != nil { + return lc, err + } + lease, cancel := context.WithCancel(lc.Context()) + select { + case l.lost <- cancel: + default: + panic("too many lock attempts") + } + return LockContext{ctx: lease, cancel: cancel}, nil +} + +// getNSLocks never holds one lock while it waits for another for long: it +// releases what it holds and retries, also after a lost lease, until its +// timeout or the request ends, and leaves nothing behind. +func TestGetNSLocksTransactional(t *testing.T) { + ns := newNSLock(false) + lock := func(name string) RWLocker { return ns.NewNSLock(nil, "bucket", name) } + newLock := func(name string) func() RWLocker { return func() RWLocker { return lock(name) } } + reqs := func() []nsLockRequest { + return []nsLockRequest{{newLock: newLock("a"), write: true, atomic: true}, {newLock: newLock("b"), write: true, atomic: true}} + } + + t.Run("partial", func(t *testing.T) { + releaseB := ldHoldLock(t, lock("b")) + defer releaseB() + type result struct { + unlock func() + err error + } + done := make(chan result, 1) + go func() { + _, unlock, err := getNSLocks(t.Context(), reqs(), globalOperationTimeout) + done <- result{unlock, err} + }() + // "a" is taken while "b" is waited for, and free again between + // attempts. + ldTaken(t, lock("a")) + deadline := time.Now().Add(3 * nsLockTryTimeout) + for !ldTry(t, lock("a"), true) { + if time.Now().After(deadline) { + t.Fatal("a stays held while b is waited for") + } + } + releaseB() + res := <-done + if res.err != nil { + t.Fatal(res.err) + } + if ldTry(t, lock("a"), false) || ldTry(t, lock("b"), false) { + t.Error("not both locks held") + } + res.unlock() + if !ldTry(t, lock("a"), true) || !ldTry(t, lock("b"), true) { + t.Error("locks left behind") + } + }) + + t.Run("timeout", func(t *testing.T) { + releaseB := ldHoldLock(t, lock("b")) + defer releaseB() + start := time.Now() + _, _, err := getNSLocks(t.Context(), reqs(), newDynamicTimeout(3*time.Second, 3*time.Second)) + if !errors.As(err, &OperationTimedOut{}) { + t.Fatalf("got %v, want a timeout", err) + } + if took := time.Since(start); took > 6*time.Second { + t.Errorf("timed out after %v", took) + } + if !ldTry(t, lock("a"), true) { + t.Error("a left behind") + } + }) + + t.Run("cancel", func(t *testing.T) { + releaseB := ldHoldLock(t, lock("b")) + defer releaseB() + ctx, cancel := context.WithCancel(t.Context()) + time.AfterFunc(300*time.Millisecond, cancel) + _, _, err := getNSLocks(ctx, reqs(), globalOperationTimeout) + if !errors.Is(err, context.Canceled) { + t.Fatalf("got %v, want canceled", err) + } + if !ldTry(t, lock("a"), true) { + t.Error("a left behind") + } + }) + + t.Run("lease", func(t *testing.T) { + lost := make(chan context.CancelFunc, 64) + leased := []nsLockRequest{{newLock: func() RWLocker { return ldLeaseLock{lock("a"), lost} }, write: true, atomic: true}, {newLock: newLock("b"), write: true, atomic: true}} + // Losing the first lease while the second is waited for retries + // with a fresh context. + releaseB := ldHoldLock(t, lock("b")) + type result struct { + lkctx LockContext + unlock func() + err error + } + done := make(chan result, 1) + go func() { + lkctx, unlock, err := getNSLocks(t.Context(), leased, globalOperationTimeout) + done <- result{lkctx, unlock, err} + }() + (<-lost)() + releaseB() + res := <-done + if res.err != nil { + t.Fatal(res.err) + } + if res.lkctx.Context().Err() != nil { + t.Fatal("returned context ended before any lease was lost") + } + // Losing a held lease ends the returned context. + var cancel context.CancelFunc + for len(lost) > 0 { + cancel = <-lost + } + cancel() + select { + case <-res.lkctx.Context().Done(): + case <-time.After(time.Second): + t.Error("lost lease did not end the locks' context") + } + res.unlock() + if !ldTry(t, lock("a"), true) || !ldTry(t, lock("b"), true) { + t.Error("locks left behind") + } + }) +} + +// A group whose lock times out fails alone and the later groups still run. +func TestDeleteObjectsInSetsLaterGroupsRun(t *testing.T) { + for _, mode := range ldModes[1:] { + t.Run(mode, func(t *testing.T) { + z, bucket := ldPool(t, mode) + k0, k1 := ldKey(t, z, "later", 0), ldKey(t, z, "later", 1) + ldMustPut(t, z, bucket, k0, "zero") + ldMustPut(t, z, bucket, k1, "one") + release := ldHoldLock(t, z.serverPools[0].getHashedSet(k0).NewNSLock(bucket, k0)) + defer release() + objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: k0}}, {ObjectV: ObjectV{ObjectName: k1}}} + dobjects := []DeletedObject{{ObjectName: k0}, {ObjectName: k1}} + derrs := make([]error, 2) + z.deleteObjectsInSets(t.Context(), bucket, objects, ObjectOptions{}, dobjects, derrs, newDynamicTimeout(time.Second, time.Second)) + if !errors.As(derrs[0], &OperationTimedOut{}) || derrs[1] != nil { + t.Fatalf("results %v, want a timeout for k0 only", derrs) + } + release() + if body, _ := ldBody(t, z, bucket, k0); body != "zero" { + t.Errorf("k0 = %q", body) + } + ldAbsent(t, z, bucket, k1) + }) + } +} + +// The local lock takes the names of a group one by one. A group that waits +// for a name must not keep the names it holds: a part upload holds its part +// lock and then reads its upload ID, which the batch names too. +func TestDeleteObjectsInSetsLocalGroupReleases(t *testing.T) { + z, bucket := ldPool(t, ldLocal) + pool := z.serverPools[0] + var upload, part string + for i := range 1000 { + upload = ldKey(t, z, "local-upload", 1) + fmt.Sprintf("/u%d", i) + part = upload + "/1" + if pool.getHashedSetIndex(upload) == 1 && pool.getHashedSetIndex(part) == 1 { + break + } + } + er := pool.getHashedSet(upload) + releasePart := ldHoldLock(t, er.NewNSLock(bucket, part)) + defer releasePart() + done := make(chan []error, 1) + go func() { + _, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: upload}}, {ObjectV: ObjectV{ObjectName: part}}}, ObjectOptions{}) + done <- errs + }() + // The batch takes the upload name, waits for the part, and lets the + // part upload's read of its upload ID through. + ldTaken(t, er.NewNSLock(bucket, upload)) + read := er.NewNSLock(bucket, upload) + rctx, err := read.GetRLock(t.Context(), newDynamicTimeout(4*nsLockTryTimeout, 4*nsLockTryTimeout)) + if err != nil { + t.Fatalf("batch kept a name while waiting: %v", err) + } + read.RUnlock(rctx) + releasePart() + select { + case errs := <-done: + if errors.Join(errs...) != nil { + t.Fatal(errs) + } + case <-time.After(30 * time.Second): + t.Fatal("bulk delete did not finish") + } +} + +// ldLostUnlockReply applies its first unlock but loses the reply until gate +// opens, so dsync retries that unlock in the background. +type ldLostUnlockReply struct { + dsync.NetLocker + gate <-chan struct{} + applied, retried chan<- struct{} + calls atomic.Int32 +} + +func (l *ldLostUnlockReply) Unlock(ctx context.Context, args dsync.LockArgs) (bool, error) { + n := l.calls.Add(1) + ok, err := l.NetLocker.Unlock(ctx, args) + switch n { + case 1: + l.applied <- struct{}{} + <-l.gate + return false, errors.New("unlock reply lost") + case 2: + l.retried <- struct{}{} + } + return ok, err +} + +// ldFailFirst fails the first acquisition of any of its instances. +type ldFailFirst struct { + RWLocker + calls *atomic.Int32 +} + +func (l ldFailFirst) GetLock(ctx context.Context, timeout *dynamicTimeout) (LockContext, error) { + if l.calls.Add(1) == 1 { + return LockContext{}, OperationTimedOut{} + } + return l.RWLocker.GetLock(ctx, timeout) +} + +// A distributed unlock of a failed attempt that is retried late must not +// release the lock the next attempt took. +func TestGetNSLocksFreshLockPerAttempt(t *testing.T) { + gate := make(chan struct{}) + applied, retried := make(chan struct{}, 3), make(chan struct{}, 3) + lockers := make([]dsync.NetLocker, 3) + for i := range lockers { + lockers[i] = &ldLostUnlockReply{NetLocker: newLocker(), gate: gate, applied: applied, retried: retried} + } + ns := newNSLock(true) + dst := func() RWLocker { + return ns.NewNSLock(func() ([]dsync.NetLocker, string) { return lockers, "test" }, "bucket", "destination") + } + local := newNSLock(false) + var calls atomic.Int32 + src := func() RWLocker { return ldFailFirst{local.NewNSLock(nil, "bucket", "source"), &calls} } + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + opened := false + open := func() { + if !opened { + opened = true + close(gate) + } + } + defer open() + _, unlock, err := getNSLocks(ctx, []nsLockRequest{{newLock: dst, write: true, atomic: true}, {newLock: src, write: true, atomic: true}}, newDynamicTimeout(10*time.Second, 10*time.Second)) + if err != nil { + t.Fatal(err) + } + defer unlock() + for range lockers { + select { + case <-applied: + case <-ctx.Done(): + t.Fatal("the first attempt's unlock never reached the lockers") + } + } + open() + for range lockers { + select { + case <-retried: + case <-ctx.Done(): + t.Fatal("the lost unlock was not retried") + } + } + competitor := dst() + lc, err := competitor.GetLock(ctx, newDynamicTimeout(300*time.Millisecond, 300*time.Millisecond)) + if err == nil { + competitor.Unlock(lc) + t.Fatal("a late unlock retry released the held lock") + } +} + +// One attempt at a local lock over several names is bounded as a whole, and so +// is the acquisition. +func TestGetNSLocksLocalAttemptBound(t *testing.T) { + ns := newNSLock(false) + a, b := ns.NewNSLock(nil, "bucket", "a"), ns.NewNSLock(nil, "bucket", "b") + releaseA, releaseB := ldHoldLock(t, a), ldHoldLock(t, b) + defer releaseA() + defer releaseB() + time.AfterFunc(900*time.Millisecond, releaseA) + time.AfterFunc(1800*time.Millisecond, releaseB) + start := time.Now() + _, unlock, err := getNSLocks(t.Context(), []nsLockRequest{{ + newLock: func() RWLocker { return ns.NewNSLock(nil, "bucket", "a", "b") }, write: true, + }}, newDynamicTimeout(1500*time.Millisecond, 1500*time.Millisecond)) + if err == nil { + unlock() + t.Fatalf("acquired after %v with a deadline of 1.5s", time.Since(start)) + } + if !errors.As(err, &OperationTimedOut{}) { + t.Fatalf("got %v, want a timeout", err) + } +} + +// ldLeaseLayer gives the COPY handler locks whose lease the test ends, and +// holds its destination lookup until the lookup's context ends. +type ldLeaseLayer struct { + ObjectLayer + destination string + lost chan context.CancelFunc + lookup chan context.Context +} + +func (l *ldLeaseLayer) lockObjects(ctx context.Context, reqs ...objectLockRequest) (LockContext, func(), error) { + lc, unlock, err := l.ObjectLayer.lockObjects(ctx, reqs...) + if err != nil { + return lc, unlock, err + } + lease, cancel := context.WithCancel(lc.Context()) + l.lost <- cancel + return LockContext{ctx: lease, cancel: cancel}, func() { cancel(); unlock() }, nil +} + +func (l *ldLeaseLayer) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + if object == l.destination && opts.NoLock { + l.lookup <- ctx + <-ctx.Done() + return ObjectInfo{}, ctx.Err() + } + return l.ObjectLayer.GetObjectInfo(ctx, bucket, object, opts) +} + +// The COPY handler's work under its locks ends when a lease is lost, the +// destination lookups included. +func TestLockDomainCopyLookupEndsWithLease(t *testing.T) { + z, bucket, _ := ldAPI(t, ldShared) + ldMustPut(t, z, bucket, "source", "") + prevTier := globalTierConfigMgr + t.Cleanup(func() { globalTierConfigMgr = prevTier }) + globalTierConfigMgr = NewTierConfigMgr() + globalTierConfigMgr.Tiers["WARM"] = madmin.TierConfig{Version: madmin.TierConfigVer, Type: madmin.S3, Name: "WARM", S3: &madmin.TierS3{Bucket: "warm"}} + layer := &ldLeaseLayer{ObjectLayer: z, destination: "destination", lost: make(chan context.CancelFunc, 1), lookup: make(chan context.Context, 1)} + router := initTestAPIEndPoints(layer, []string{"CopyObject"}) + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + code := make(chan int, 1) + go func() { code <- ldCopyRequest(ctx, t, router, bucket, "source", "destination", nil).Code }() + var lookup context.Context + select { + case lookup = <-layer.lookup: + case <-ctx.Done(): + t.Fatal("COPY did not look up its destination") + } + (<-layer.lost)() + select { + case <-lookup.Done(): + case <-time.After(5 * time.Second): + t.Fatal("the destination lookup outlived the lost lease") + } + if c := <-code; c == http.StatusOK { + t.Error("COPY succeeded after losing its lock") + } +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index c168e22f9..221156480 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1239,8 +1239,26 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob object = encodeDirObject(object) } - // Serialize logical deletion with pooled writes and metadata updates. - lk := z.NewNSLock(bucket, object) + // Serialize logical deletion with the writes and metadata updates of the + // object: take the lock they take (see NewObjectNSLock). With one pool + // that is the object's own erasure set, which is where PutObject, + // multipart commit and the metadata updates lock, so a pool whose sets + // use different hosts still excludes them. With several pools it is the + // pools lock, as before. Exact-key deletes that use prefix deletion + // (DeletePrefixObject) are deletes of one object. A true prefix deletion + // keeps the pools lock on the prefix name, which never covered the + // objects below it. + // + // Nodes that still run a release before this routing lock single-pool + // deletes and copies in the first set. In a pool whose sets use different + // hosts they do not exclude the deletes and copies of updated nodes, nor + // their writes, so such a pool must not run mixed versions. + var lk RWLocker + if opts.DeletePrefix && !opts.DeletePrefixObject { + lk = z.NewNSLock(bucket, object) + } else { + lk = z.NewObjectNSLock(bucket, object) + } lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout) if err != nil { return ObjectInfo{}, err @@ -1322,10 +1340,10 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob // Evaluate the conditional (If-Match) precondition while the write lock // acquired above is held, before the delete-marker short-circuit and before - // any version is removed, so the object cannot change between the check and - // the delete. This is scoped to a single erasure set (see the note at the - // lock above): only there do the delete lock and the write path share the - // same lock namespace, making the check-then-delete atomic. + // any version is removed. The object's writers take the same lock (see the + // note at the lock above), so the object cannot change between the check + // and the delete. Multi-pool conditional deletes return above, through + // deleteObjectReconciled, under the same pools lock. if opts.CheckPrecondFn != nil { // pinfo.ObjInfo is the current latest version. getPoolInfoExistingWithOpts // intentionally strips VersionID, so for a version-scoped delete read the @@ -1509,6 +1527,18 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o objSets.Add(objects[i].ObjectName) } + // A single pool locks each name in the set its writers lock it in, one + // set at a time. A single node keeps one lock table for all sets, so it + // keeps the one bulk lock below: its local lock takes the names one by + // one and holds the earlier names while it waits. + if z.SinglePool() && !z.serverPools[0].sharesLocalNSLock() { + z.deleteObjectsInSets(ctx, bucket, objects, opts, dobjects, derrs, globalOperationTimeout) + for i := range dobjects { + dobjects[i].ObjectName = decodeDirObject(dobjects[i].ObjectName) + } + return dobjects, derrs + } + // Acquire a bulk write lock across 'objects' multiDeleteLock := z.NewNSLock(bucket, objSets.ToSlice()...) lkctx, err := multiDeleteLock.GetLock(ctx, globalOperationTimeout) @@ -1633,8 +1663,10 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec cpSrcDstSame := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(dstBucket, dstObject)) + // A caller that also locks the source takes both with lockObjects, before + // it reads the source, and passes NoLock. if !dstOpts.NoLock { - ns := z.NewNSLock(dstBucket, dstObject) + ns := z.NewObjectNSLock(dstBucket, dstObject) lkctx, err := ns.GetLock(ctx, globalOperationTimeout) if err != nil { return ObjectInfo{}, err @@ -2231,7 +2263,7 @@ func (z *erasureServerPools) GetMultipartInfo(ctx context.Context, bucket, objec return MultipartInfo{}, err } - uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID)) + uploadIDLock := z.uploadIDNSLock(bucket, object, uploadID) lkctx, err := uploadIDLock.GetRLock(ctx, globalOperationTimeout) if err != nil { return MultipartInfo{}, err @@ -2272,7 +2304,7 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object return ListPartsInfo{}, err } - uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID)) + uploadIDLock := z.uploadIDNSLock(bucket, object, uploadID) lkctx, err := uploadIDLock.GetRLock(ctx, globalOperationTimeout) if err != nil { return ListPartsInfo{}, err @@ -2323,7 +2355,7 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o } }(ctx) - lk := z.NewNSLock(bucket, pathJoin(object, uploadID)) + lk := z.uploadIDNSLock(bucket, object, uploadID) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) if err != nil { return err @@ -2363,12 +2395,10 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket // Keep object -> upload lock ordering for single-pool completion too: // native restore holds the object lock while publishing upload parts. + // Both locks are in the object's lock domain, the set-local one of + // single-pool writes. 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)) - } + lk := z.NewObjectNSLock(bucket, encodeDirObject(object)) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) if err != nil { return objInfo, err @@ -2433,7 +2463,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket // An empty upload ID aliases the object lock already held above. // Let the storage layer return its normal missing bucket/upload error. if uploadID != "" { - uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID)) + uploadIDLock := z.uploadIDNSLock(bucket, object, uploadID) wlkctx, err := uploadIDLock.GetLock(ctx, globalOperationTimeout) if err != nil { return objInfo, err diff --git a/cmd/metacache-server-pool.go b/cmd/metacache-server-pool.go index 2889e9a96..276cef0fb 100644 --- a/cmd/metacache-server-pool.go +++ b/cmd/metacache-server-pool.go @@ -265,9 +265,12 @@ func (z *erasureServerPools) listPathShortcutEOF(ctx context.Context, bucket str } // listingRLock covers the lock domains used by normal same-key mutations. -// Multi-pool writes use the pools lock. In a single pool, DELETE/COPY use the -// first set while PUT and metadata updates use the destination set, so recovery -// takes both in that order. No writer routing or wire protocol changes here. +// Multi-pool writes use the pools lock. In a single pool every object write, +// DELETE and COPY included, locks the object's own set (see NewObjectNSLock), +// which is the source set here. Releases before that routing locked +// single-pool DELETE and COPY in the first set, so recovery still takes the +// first set too, in that order, while nodes of those releases may run. It can +// drop that lock only behind a version boundary that keeps such nodes out. func (z *erasureServerPools) listingRLock(ctx context.Context, bucket, object string, source *erasureObjects) (LockContext, func(), error) { firstSet := z.serverPools[0].sets[0] first := z.NewNSLock(bucket, object) diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 0f4a663ac..11e867171 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -262,7 +262,13 @@ func (o *ObjectOptions) SetEvalRetentionBypassFn(f EvalRetentionBypassFn) { // ObjectLayer implements primitives for object API layer. type ObjectLayer interface { // Locking operations on object. + // NewNSLock locks names in one stable domain. It is for names that are + // not objects, such as bucket and IAM metadata locks. NewNSLock(bucket string, objects ...string) RWLocker + // NewObjectNSLock locks one object in the lock domain of its writes. + NewObjectNSLock(bucket, object string) RWLocker + // lockObjects locks several objects together, see erasureServerPools. + lockObjects(ctx context.Context, reqs ...objectLockRequest) (LockContext, func(), error) // Storage operations. Shutdown(context.Context) error diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 81af341f5..a1133be7e 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -34,6 +34,7 @@ import ( "sort" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -183,7 +184,7 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r // Take read lock on object, here so subsequent lower-level // calls do not need to. - lock := objectAPI.NewNSLock(bucket, object) + lock := objectAPI.NewObjectNSLock(bucket, object) lkctx, err := lock.GetRLock(ctx, globalOperationTimeout) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) @@ -1470,12 +1471,35 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re return checkCopyObjectPreconditions(ctx, w, r, o) } getOpts.CheckPrecondFn = checkCopyPrecondFn - if cpSrcDstSame { - getOpts.NoLock = true + + // Lock the source for reading and a local destination for writing before + // the source is opened, both in the lock domain of their writers. Taking + // them together in one order keeps copies in opposite directions from + // holding each other; a copy onto itself takes one write lock. The source + // is then read, preconditions and inline data included, and the + // destination looked up and written, all under these locks and without + // taking them again. That work runs in lockCtx, which ends when either + // lock is lost and so aborts the copy; errors are still reported in the + // request's context. Both locks are released once the copy has been + // written: a synchronous replication below reads the destination with its + // own lock. + locks := []objectLockRequest{{bucket: srcBucket, object: srcObject}} + if !remoteCallRequired { + locks = append(locks, objectLockRequest{bucket: dstBucket, object: dstObject, write: true}) } + lkctx, unlock, err := objectAPI.lockObjects(ctx, locks...) + if err != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) + return + } + unlockCopy := sync.OnceFunc(unlock) + defer unlockCopy() + lockCtx := lkctx.Context() + getOpts.NoLock = true + dstOpts.NoLock = true var rs *HTTPRangeSpec - gr, err := getObjectNInfo(ctx, srcBucket, srcObject, rs, r.Header, getOpts) + gr, err := getObjectNInfo(lockCtx, srcBucket, srcObject, rs, r.Header, getOpts) if err != nil { if isErrPreconditionFailed(err) { return @@ -1542,7 +1566,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re length := actualSize if !cpSrcDstSame { - if err := enforceBucketQuotaHard(ctx, dstBucket, actualSize); err != nil { + if err := enforceBucketQuotaHard(lockCtx, dstBucket, actualSize); err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return } @@ -1560,12 +1584,12 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re compressMetadata[ReservedMetadataPrefix+"compression"] = compressionAlgorithmV2 compressMetadata[ReservedMetadataPrefix+"actual-size"] = strconv.FormatInt(actualSize, 10) - reader = etag.NewReader(ctx, reader, nil, nil) + reader = etag.NewReader(lockCtx, reader, nil, nil) } else { reader = gr } - srcInfo.Reader, err = hash.NewReader(ctx, reader, length, "", "", actualSize) + srcInfo.Reader, err = hash.NewReader(lockCtx, reader, length, "", "", actualSize) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return @@ -1660,7 +1684,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re } } - if err = rotateKey(ctx, oldKey, newKeyID, newKey, srcBucket, srcObject, encMetadata, kmsCtx); err != nil { + if err = rotateKey(lockCtx, oldKey, newKeyID, newKey, srcBucket, srcObject, encMetadata, kmsCtx); err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return } @@ -1745,7 +1769,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re dstOpts.IndexCB = cb defer s2c.Close() reader = etag.Wrap(s2c, checksumReader) - srcInfo.Reader, err = hash.NewReader(ctx, reader, -1, "", "", actualSize) + srcInfo.Reader, err = hash.NewReader(lockCtx, reader, -1, "", "", actualSize) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return @@ -1758,7 +1782,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re if isTargetEncrypted && !remoteCallRequired { var encReader io.Reader kind, _ := crypto.IsRequested(r.Header) - encReader, objEncKey, err = newEncryptReader(ctx, srcInfo.Reader, kind, newKeyID, newKey, dstBucket, dstObject, encMetadata, kmsCtx) + encReader, objEncKey, err = newEncryptReader(lockCtx, srcInfo.Reader, kind, newKeyID, newKey, dstBucket, dstObject, encMetadata, kmsCtx) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return @@ -1773,7 +1797,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re } // do not try to verify encrypted content - srcInfo.Reader, err = hash.NewReader(ctx, reader, targetSize, "", "", actualSize) + srcInfo.Reader, err = hash.NewReader(lockCtx, reader, targetSize, "", "", actualSize) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return @@ -1845,7 +1869,12 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re srcInfo.UserDefined = objectlock.FilterObjectLockMetadata(srcInfo.UserDefined, true, true) retPerms := isPutActionAllowed(ctx, getRequestAuthType(r), dstBucket, dstObject, r, policy.PutObjectRetentionAction) holdPerms := isPutActionAllowed(ctx, getRequestAuthType(r), dstBucket, dstObject, r, policy.PutObjectLegalHoldAction) - getObjectInfo := objectAPI.GetObjectInfo + // The destination is locked above: look it up without its lock, in the + // locks' context. + getObjectInfo := func(_ context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + opts.NoLock = true + return objectAPI.GetObjectInfo(lockCtx, bucket, object, opts) + } // apply default bucket configuration/governance headers for dest side. retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted) @@ -2031,9 +2060,11 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re // For an ordinary federated copy, actualSize is the logical plaintext // length to declare. srcInfo.Size is not a reliable wire length because // the read path may already have adjusted it. - writeCtx, modified := withFederatedWriteTime(ctx) + writeCtx, modified := withFederatedWriteTime(lockCtx) remoteObjInfo, rerr := core.PutObject(writeCtx, dstBucket, dstObject, srcInfo.Reader, actualSize, "", "", opts) + gr.Close() + unlockCopy() if rerr != nil { writeErrorResponse(ctx, w, toAPIError(ctx, rerr), r.URL) return @@ -2080,7 +2111,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re // Copy source object to destination, if source and destination // object is same then only metadata is updated. - objInfo, err = copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts) + objInfo, err = copyObjectFn(lockCtx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts) + gr.Close() + unlockCopy() if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return diff --git a/cmd/object-retention-authz_test.go b/cmd/object-retention-authz_test.go index af44aac37..c62477dc0 100644 --- a/cmd/object-retention-authz_test.go +++ b/cmd/object-retention-authz_test.go @@ -544,6 +544,18 @@ func (p *hookBeforeWrite) CompleteMultipartUpload(ctx context.Context, bucket, k return p.ObjectLayer.CompleteMultipartUpload(ctx, bucket, key, uploadID, parts, opts) } +// The COPY handler checks Object Lock under its destination's write lock, so +// no writer can land between that check and the copy. The latest a writer can +// land is just before the handler takes its locks. +func (p *hookBeforeWrite) lockObjects(ctx context.Context, reqs ...objectLockRequest) (LockContext, func(), error) { + for _, r := range reqs { + if r.write { + p.fire(r.object) + } + } + return p.ObjectLayer.lockObjects(ctx, reqs...) +} + func TestDirectoryObjectLockAfterPreflight(t *testing.T) { ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ t: t,