From fd3b437e04d55136e0d780d44402dc4709c4a98f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 06:57:02 -0700 Subject: [PATCH] s3api: write the null version's object-lock metadata to the plain entry (#12033) * s3api: write the null version's lock metadata to the plain entry PutObjectRetention and PutObjectLegalHold with versionId=null loaded the pre-versioning entry but wrote the update under .versions/v_null, so the acknowledged lock never bound the object it was read from. Resolve the entry and its write path in one helper that applies the same null-version rule as getSpecificObjectVersion. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3api: cover the null-version lock write path Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3api: persist object-lock metadata with an in-place entry update mkFile rebuilds the target entry from chunks alone, so a lock write on a plain (null-version) entry discarded attributes, inline content, and the remote reference. UpdateEntry keeps the whole fetched entry, matching how the tagging and ACL writers persist metadata. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/s3api/s3api_object_retention.go | 163 ++++++++-------------- weed/s3api/s3api_object_retention_test.go | 100 +++++++++++++ 2 files changed, 160 insertions(+), 103 deletions(-) diff --git a/weed/s3api/s3api_object_retention.go b/weed/s3api/s3api_object_retention.go index 2d1e8f4ff2..53f92f3574 100644 --- a/weed/s3api/s3api_object_retention.go +++ b/weed/s3api/s3api_object_retention.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/util" ) // ==================================================================== @@ -213,6 +214,49 @@ func (s3a *S3ApiServer) getObjectEntry(bucket, object, versionId string) (*filer return entry, nil } +// getObjectEntryForUpdate resolves the entry a metadata update targets and the +// filer path it is written under. It applies the same version resolution as +// getSpecificObjectVersion - "null" names the plain pre-versioning entry, not a +// .versions file - so the write lands on the entry that was read. +func (s3a *S3ApiServer) getObjectEntryForUpdate(bucket, object, versionId string) (entry *filer_pb.Entry, entryPath string, err error) { + entryPath = object + if versionId != "" { + entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId) + if err != nil { + return nil, "", fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, ErrVersionNotFound) + } + if versionId != "null" { + entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) + } + return entry, entryPath, nil + } + + versioningEnabled, vErr := s3a.isVersioningEnabled(bucket) + if vErr != nil { + return nil, "", fmt.Errorf("error checking versioning: %w", vErr) + } + if !versioningEnabled { + entry, err = s3a.fetchObjectEntryRequired(bucket, object) + if err != nil { + return nil, "", fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound) + } + return entry, entryPath, nil + } + + entry, err = s3a.getLatestObjectVersion(bucket, object) + if err != nil { + return nil, "", fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, ErrLatestVersionNotFound) + } + if entry.Extended != nil { + if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists { + if storedVersionId := string(versionIdBytes); storedVersionId != "null" && isValidVersionID(storedVersionId) { + entryPath = object + ".versions/" + s3a.getVersionFileName(storedVersionId) + } + } + } + return entry, entryPath, nil +} + // ==================================================================== // RETENTION OPERATIONS // ==================================================================== @@ -254,47 +298,9 @@ func (s3a *S3ApiServer) getObjectRetention(bucket, object, versionId string) (*O // setObjectRetention sets object retention configuration func (s3a *S3ApiServer) setObjectRetention(bucket, object, versionId string, retention *ObjectRetention, bypassGovernance bool) error { - var entry *filer_pb.Entry - var err error - var entryPath string - - if versionId != "" { - entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId) - if err != nil { - return fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, ErrVersionNotFound) - } - entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) - } else { - // Check if versioning is enabled - versioningEnabled, vErr := s3a.isVersioningEnabled(bucket) - if vErr != nil { - return fmt.Errorf("error checking versioning: %w", vErr) - } - - if versioningEnabled { - entry, err = s3a.getLatestObjectVersion(bucket, object) - if err != nil { - return fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, ErrLatestVersionNotFound) - } - // Extract version ID from entry metadata - entryPath = object // default to regular object path - if entry.Extended != nil { - if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists { - versionId = string(versionIdBytes) - // A stored version id must be a valid path segment before it can - // name a .versions file; otherwise fall back to the regular object. - if versionId != "null" && isValidVersionID(versionId) { - entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) - } - } - } - } else { - entry, err = s3a.fetchObjectEntryRequired(bucket, object) - if err != nil { - return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound) - } - entryPath = object - } + entry, entryPath, err := s3a.getObjectEntryForUpdate(bucket, object, versionId) + if err != nil { + return err } // Check if object is already under retention @@ -355,18 +361,15 @@ func (s3a *S3ApiServer) setObjectRetention(bucket, object, versionId string, ret entry.WormEnforcedAtTsNs = time.Now().UnixNano() } - // Update the entry + // Update the entry in place. UpdateEntry persists the whole fetched entry, so + // attributes, inline content, and remote references survive the metadata change + // where mkFile's fresh entry would drop them. // NOTE: Potential race condition exists if concurrent calls to PutObjectRetention // and PutObjectLegalHold update the same object simultaneously, as they might - // overwrite each other's Extended map changes. This is mitigated by the fact - // that mkFile operations are typically serialized at the filer level, but - // future implementations might consider using atomic update operations or - // entry-level locking for complete safety. - bucketDir := s3a.bucketDir(bucket) - return s3a.mkFile(bucketDir, entryPath, entry.Chunks, func(updatedEntry *filer_pb.Entry) { - updatedEntry.Extended = entry.Extended - updatedEntry.WormEnforcedAtTsNs = entry.WormEnforcedAtTsNs - }) + // overwrite each other's Extended map changes. + dir, name := util.NewFullPath(s3a.bucketDir(bucket), entryPath).DirAndName() + entry.Name = name + return s3a.updateEntry(dir, entry) } // ==================================================================== @@ -399,47 +402,9 @@ func (s3a *S3ApiServer) getObjectLegalHold(bucket, object, versionId string) (*O // setObjectLegalHold sets object legal hold configuration func (s3a *S3ApiServer) setObjectLegalHold(bucket, object, versionId string, legalHold *ObjectLegalHold) error { - var entry *filer_pb.Entry - var err error - var entryPath string - - if versionId != "" { - entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId) - if err != nil { - return fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, ErrVersionNotFound) - } - entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) - } else { - // Check if versioning is enabled - versioningEnabled, vErr := s3a.isVersioningEnabled(bucket) - if vErr != nil { - return fmt.Errorf("error checking versioning: %w", vErr) - } - - if versioningEnabled { - entry, err = s3a.getLatestObjectVersion(bucket, object) - if err != nil { - return fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, ErrLatestVersionNotFound) - } - // Extract version ID from entry metadata - entryPath = object // default to regular object path - if entry.Extended != nil { - if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists { - versionId = string(versionIdBytes) - // A stored version id must be a valid path segment before it can - // name a .versions file; otherwise fall back to the regular object. - if versionId != "null" && isValidVersionID(versionId) { - entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) - } - } - } - } else { - entry, err = s3a.fetchObjectEntryRequired(bucket, object) - if err != nil { - return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound) - } - entryPath = object - } + entry, entryPath, err := s3a.getObjectEntryForUpdate(bucket, object, versionId) + if err != nil { + return err } // Update legal hold metadata @@ -449,17 +414,9 @@ func (s3a *S3ApiServer) setObjectLegalHold(bucket, object, versionId string, leg entry.Extended[s3_constants.ExtLegalHoldKey] = []byte(legalHold.Status) - // Update the entry - // NOTE: Potential race condition exists if concurrent calls to PutObjectRetention - // and PutObjectLegalHold update the same object simultaneously, as they might - // overwrite each other's Extended map changes. This is mitigated by the fact - // that mkFile operations are typically serialized at the filer level, but - // future implementations might consider using atomic update operations or - // entry-level locking for complete safety. - bucketDir := s3a.bucketDir(bucket) - return s3a.mkFile(bucketDir, entryPath, entry.Chunks, func(updatedEntry *filer_pb.Entry) { - updatedEntry.Extended = entry.Extended - }) + dir, name := util.NewFullPath(s3a.bucketDir(bucket), entryPath).DirAndName() + entry.Name = name + return s3a.updateEntry(dir, entry) } // ==================================================================== diff --git a/weed/s3api/s3api_object_retention_test.go b/weed/s3api/s3api_object_retention_test.go index 44089deb33..201d471658 100644 --- a/weed/s3api/s3api_object_retention_test.go +++ b/weed/s3api/s3api_object_retention_test.go @@ -1,15 +1,23 @@ package s3api import ( + "context" "fmt" "io" "net/http" "strings" + "sync" "testing" "time" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/wdclient" ) func TestValidateRetention(t *testing.T) { @@ -857,3 +865,95 @@ func TestStoreObjectLockConfigurationClearsStaleYears(t *testing.T) { func timePtr(t time.Time) *time.Time { return &t } + +// retentionTestFiler serves the two filer calls a lock write needs: the lookup +// that resolves the version and the update that persists the modified entry. +type retentionTestFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + + mu sync.Mutex + entries map[string]*filer_pb.Entry + updates []string +} + +func (f *retentionTestFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok { + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}, nil + } + return &filer_pb.LookupDirectoryEntryResponse{}, nil +} + +func (f *retentionTestFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.updates = append(f.updates, req.Directory+"/"+req.Entry.Name) + f.entries[req.Directory+"/"+req.Entry.Name] = req.Entry + return &filer_pb.UpdateEntryResponse{}, nil +} + +func newRetentionTestServer(t *testing.T, f *retentionTestFiler) *S3ApiServer { + t.Helper() + addr := startFakeFiler(t, f) + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + return &S3ApiServer{ + option: &S3ApiServerOption{ + Filers: []pb.ServerAddress{addr}, + GrpcDialOption: dialOption, + BucketsPath: "/buckets", + }, + filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""), + } +} + +// A "null" version id resolves to the pre-versioning entry under the plain +// object key, the same place getSpecificObjectVersion reads it from. Writing +// the lock under .versions/v_null leaves the real entry unlocked. +func TestNullVersionLockWritesToPlainEntry(t *testing.T) { + f := &retentionTestFiler{entries: map[string]*filer_pb.Entry{ + "/buckets/b/key": { + Name: "key", + Chunks: []*filer_pb.FileChunk{{FileId: "1,0000000000000001"}}, + Content: []byte("inline"), + Attributes: &filer_pb.FuseAttributes{ + Mime: "text/plain", + Md5: []byte("etagmd5"), + }, + Extended: map[string][]byte{ + "ETag": []byte("\"etagmd5\""), + }, + }, + }} + s3a := newRetentionTestServer(t, f) + + retainUntil := time.Now().Add(24 * time.Hour).Truncate(time.Second) + err := s3a.setObjectRetention("b", "key", "null", &ObjectRetention{ + Mode: s3_constants.RetentionModeCompliance, + RetainUntilDate: &retainUntil, + }, false) + if err != nil { + t.Fatalf("setObjectRetention: %v", err) + } + if len(f.updates) != 1 || f.updates[0] != "/buckets/b/key" { + t.Fatalf("retention write landed on %v, want /buckets/b/key", f.updates) + } + updated := f.entries["/buckets/b/key"] + if got := string(updated.Extended[s3_constants.ExtObjectLockModeKey]); got != s3_constants.RetentionModeCompliance { + t.Fatalf("plain entry lock mode = %q, want COMPLIANCE", got) + } + if updated.Attributes.Mime != "text/plain" || len(updated.Content) == 0 || len(updated.Chunks) != 1 || string(updated.Extended["ETag"]) != "\"etagmd5\"" { + t.Fatalf("lock write dropped entry fields: %+v", updated) + } + + f.updates = nil + if err := s3a.setObjectLegalHold("b", "key", "null", &ObjectLegalHold{Status: "ON"}); err != nil { + t.Fatalf("setObjectLegalHold: %v", err) + } + if len(f.updates) != 1 || f.updates[0] != "/buckets/b/key" { + t.Fatalf("legal hold write landed on %v, want /buckets/b/key", f.updates) + } + if got := string(f.entries["/buckets/b/key"].Extended[s3_constants.ExtLegalHoldKey]); got != "ON" { + t.Fatalf("plain entry legal hold = %q, want ON", got) + } +}