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 <key>.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

@@ -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)
 }
 
 // ====================================================================

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)
+	}
+}