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,