fix: stop sidecar attributes leaking into new and promoted object versions

The `sidecar` metadata storer keeps attributes by path rather than on the file. So a new file published at an object's path inherited every attribute it didn't set itself: tags, user metadata, content headers and `x-amz-website-redirect-location`, plus legal hold and retention on version promotion. This hit overwrites by `PutObject`, `CopyObject` and `CompleteMultipartUpload`, new versions and `PutObject` over a delete marker in versioned buckets, and the version made current again by `DeleteObject` with a `versionId`. A plain `PutObject` over a multipart object also kept its `mp-metadata`, so `HeadObject` and `GetObject` with `partNumber` returned the old `PartsCount` and a `Content-Range` beyond the object size. `xattr` was not affected, since its attributes live on the inode and are replaced along with it.

Add `ReplaceObject(bucket, object, keep)` to `MetadataStorer`. It is a no-op for `XattrMeta` and `NoMeta`. `SideCar` removes the object's attribute files not named in `keep` and leaves directories in place, since they hold the metadata of objects under `<object>/meta/`. Posix calls it in `PutObject` and `CompleteMultipartUpload` after the version snapshot, and in `DeleteObject` when restoring the latest remaining version. This replaces the scattered deletes of the delete marker, `versionId` and object lock attributes. Attributes that every publish rewrites are kept and overwritten in place: the etag, the checksums, the new `versionId`, `mp-metadata` for multipart uploads, and the restored version's attributes on promotion. This way concurrent readers, such as the lock-free `If-Match` fast-fail check, never see them missing.

`PutObject` now re-checks under the publish lock that the object path hasn't become a directory while the body was staged, as `CompleteMultipartUpload` already does. This stops it from removing the attributes of a directory object created by a racing `PUT foo/`.

backend/meta/meta.go

@@ -44,6 +44,15 @@ type MetadataStorer interface {
 	// within the given bucket. This must be called whenever the data
 	// directory for an object is renamed so that metadata stays in sync.
 	RenameObject(bucket, oldObject, newObject string) error
+
+	// ReplaceObject removes the attributes of an object, other than the
+	// ones named in keep, when a new file replaces the object's file. This
+	// must be called whenever a new file is published at an object's path
+	// so that it doesn't inherit the metadata of the replaced file. The
+	// attributes in keep are the ones the new file sets anyway: they are
+	// overwritten instead of removed, so that they never go missing for
+	// concurrent readers.
+	ReplaceObject(bucket, object string, keep []string) error
 }
 
 // RootDirSetter is implemented by metadata storers that keep metadata on the

backend/meta/none.go

@@ -57,3 +57,8 @@ func (NoMeta) DeleteAttributes(bucket, object string) error {
 func (NoMeta) RenameObject(_, _, _ string) error {
 	return nil
 }
+
+// ReplaceObject is a no-op because NoMeta does not store metadata.
+func (NoMeta) ReplaceObject(_, _ string, _ []string) error {
+	return nil
+}

backend/meta/sidecar.go

@@ -20,6 +20,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"slices"
 	"syscall"
 
 	"github.com/versity/versitygw/s3err"
@@ -208,6 +209,36 @@ func (s SideCar) RenameObject(bucket, oldObject, newObject string) error {
 	return err
 }
 
+// ReplaceObject removes the attributes of the object other than the ones in
+// keep. The attributes are stored at the object's path, so they would
+// otherwise outlive the replaced file. Directories in the metadata directory
+// hold the metadata of the objects under "<object>/meta/" and are left in
+// place.
+func (s SideCar) ReplaceObject(bucket, object string, keep []string) error {
+	bucket, object = trimVolume(bucket), trimVolume(object)
+	metadir := filepath.Join(s.dir, bucket, object, sidecarmeta)
+
+	ents, err := os.ReadDir(metadir)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("failed to list attributes: %w", err)
+	}
+
+	for _, ent := range ents {
+		if ent.IsDir() || slices.Contains(keep, ent.Name()) {
+			continue
+		}
+		err := os.Remove(filepath.Join(metadir, ent.Name()))
+		if err != nil && !errors.Is(err, os.ErrNotExist) {
+			return fmt.Errorf("failed to remove attribute: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (s SideCar) cleanupEmptyDirs(metadir, bucket, object string) {
 	removeIfEmpty(metadir)
 	if bucket == "" {

backend/meta/sidecar_test.go

@@ -0,0 +1,81 @@
+// Copyright 2026 Versity Software
+// This file is licensed under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package meta
+
+import (
+	"errors"
+	"slices"
+	"testing"
+)
+
+func TestSideCarReplaceObject(t *testing.T) {
+	s, err := NewSideCar(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// no metadata stored for the object
+	if err := s.ReplaceObject("bucket", "obj", []string{"etag"}); err != nil {
+		t.Fatalf("replace object without metadata: %v", err)
+	}
+
+	for _, attr := range []string{"etag", "checksums", "X-Amz-Tagging", "legal-hold"} {
+		if err := s.StoreAttribute(nil, "bucket", "obj", attr, []byte(attr)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// the metadata of "obj/meta/child" is stored under the metadata
+	// directory of "obj"
+	if err := s.StoreAttribute(nil, "bucket", "obj/meta/child", "etag", []byte("child")); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.ReplaceObject("bucket", "obj", []string{"etag", "checksums"}); err != nil {
+		t.Fatalf("replace object: %v", err)
+	}
+
+	for _, attr := range []string{"etag", "checksums"} {
+		val, err := s.RetrieveAttribute(nil, "bucket", "obj", attr)
+		if err != nil {
+			t.Fatalf("kept attribute %v: %v", attr, err)
+		}
+		if string(val) != attr {
+			t.Errorf("kept attribute %v: expected %q, got %q", attr, attr, val)
+		}
+	}
+	for _, attr := range []string{"X-Amz-Tagging", "legal-hold"} {
+		_, err := s.RetrieveAttribute(nil, "bucket", "obj", attr)
+		if !errors.Is(err, ErrNoSuchKey) {
+			t.Errorf("removed attribute %v: expected ErrNoSuchKey, got %v", attr, err)
+		}
+	}
+
+	val, err := s.RetrieveAttribute(nil, "bucket", "obj/meta/child", "etag")
+	if err != nil {
+		t.Fatalf("nested object attribute: %v", err)
+	}
+	if string(val) != "child" {
+		t.Errorf("nested object attribute: expected %q, got %q", "child", val)
+	}
+
+	attrs, err := s.ListAttributes("bucket", "obj")
+	if err != nil {
+		t.Fatal(err)
+	}
+	slices.Sort(attrs)
+	if want := []string{"checksums", "child", "etag"}; !slices.Equal(attrs, want) {
+		t.Errorf("expected the metadata directory entries to be %v, got %v", want, attrs)
+	}
+}

backend/meta/xattr.go

@@ -143,6 +143,13 @@ func (x XattrMeta) RenameObject(_, _, _ string) error {
 	return nil
 }
 
+// ReplaceObject is a no-op for xattr because extended attributes are stored
+// on the inodes: the replaced file takes its attributes with it, and the new
+// file has only the ones set on it.
+func (x XattrMeta) ReplaceObject(_, _ string, _ []string) error {
+	return nil
+}
+
 // ListAttributes lists all attributes for an object in a bucket.
 func (x XattrMeta) ListAttributes(bucket, object string) ([]string, error) {
 	name, err := x.path(bucket, object)

backend/posix/posix.go

@@ -2886,11 +2886,16 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 		return res, "", err
 	}
 
-	// Clear the live marker after any snapshot, including when the data file
-	// is missing or versioning does not require archiving the previous object.
-	err = p.meta.DeleteAttribute(bucket, object, deleteMarkerKey)
-	if err != nil && !errors.Is(err, meta.ErrNoSuchKey) && !errors.Is(err, fs.ErrNotExist) {
-		return res, "", fmt.Errorf("delete object delete-marker: %w", err)
+	// The new object starts with none of the replaced object's attributes,
+	// whether or not they were archived above: the null version and the
+	// unversioned object are replaced without a snapshot.
+	keep := []string{etagkey, checksumsKey, mpMetaKey}
+	if p.versioningEnabled() && vEnabled {
+		keep = append(keep, versionIdKey)
+	}
+	err = p.meta.ReplaceObject(bucket, object, keep)
+	if err != nil {
+		return res, "", fmt.Errorf("remove replaced object attributes: %w", err)
 	}
 
 	// With versioning suspended the new object becomes the null version:
@@ -2900,11 +2905,6 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 		if err != nil {
 			return res, "", err
 		}
-		// Clear any stale versionId sidecar attribute left from a previous
-		// versioned object at this path.  With xattr this is implicit (the
-		// new file carries only the attrs set on the tmpfile), but with
-		// path-based metadata the old attr persists until explicitly deleted.
-		_ = p.meta.DeleteAttribute(bucket, object, versionIdKey)
 	}
 
 	objMeta := p.loadObjectMetaProperties(nil, bucket, upiddir, nil)
@@ -4439,13 +4439,6 @@ func (p *Posix) snapshotObjVersion(bucket, key string, vStatus types.BucketVersi
 		if err != nil {
 			return fmt.Errorf("create object version: %w", err)
 		}
-		// With path-based metadata backends (e.g. sidecar), object-lock
-		// attributes written on the previous version persist at this path
-		// after createObjVersion because metadata is not replaced atomically
-		// the way xattrs are on file rename.  Delete them so they do not
-		// bleed into the new version.
-		_ = p.meta.DeleteAttribute(bucket, key, objectLegalHoldKey)
-		_ = p.meta.DeleteAttribute(bucket, key, objectRetentionKey)
 	}
 
 	return nil
@@ -4586,8 +4579,7 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 		versioned := p.versioningEnabled() && vStatus != ""
 
 		// In a versioned bucket the directory is kept across versions: its
-		// current version is copied to the versioning directory and its
-		// object attributes are then replaced with the new version's.
+		// current version is copied to the versioning directory first.
 		err = p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
 		if err != nil {
 			return s3response.PutObjectOutput{}, err
@@ -4604,13 +4596,16 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, err
 		}
 
+		// The directory outlives the object it replaces, whether or not
+		// the bucket is versioned: the new object starts with none of the
+		// replaced object's attributes.
+		err = p.clearDirObjectAttrs(*po.Bucket, *po.Key)
+		if err != nil {
+			return s3response.PutObjectOutput{}, err
+		}
+
 		var versionID string
 		if versioned {
-			err = p.clearDirObjectAttrs(*po.Bucket, *po.Key)
-			if err != nil {
-				return s3response.PutObjectOutput{}, err
-			}
-
 			if p.isBucketVersioningSuspended(vStatus) {
 				err = p.deleteNullVersionIdObject(*po.Bucket, *po.Key)
 				if err != nil {
@@ -4854,6 +4849,13 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 
 	}
 
+	// The path may have become a directory while the body was staged: the
+	// attributes there belong to the directory, not to this object.
+	d, err = os.Stat(name)
+	if err == nil && d.IsDir() {
+		return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)
+	}
+
 	// Snapshot the object that is about to be replaced only after the
 	// conditional check has succeeded and while holding the publish lock.
 	verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
@@ -4861,11 +4863,16 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 		return s3response.PutObjectOutput{}, verr
 	}
 
-	// Marker cleanup belongs to publishing: snapshotting can be skipped for
-	// orphaned sidecars, suspended versioning, or a missing version ID.
-	err = p.meta.DeleteAttribute(*po.Bucket, *po.Key, deleteMarkerKey)
-	if err != nil && !errors.Is(err, meta.ErrNoSuchKey) && !errors.Is(err, fs.ErrNotExist) {
-		return s3response.PutObjectOutput{}, fmt.Errorf("delete object delete-marker: %w", err)
+	// The new object starts with none of the replaced object's attributes,
+	// whether or not they were archived above: the null version and the
+	// unversioned object are replaced without a snapshot.
+	keep := []string{etagkey, checksumsKey}
+	if versionID != "" {
+		keep = append(keep, versionIdKey)
+	}
+	err = p.meta.ReplaceObject(*po.Bucket, *po.Key, keep)
+	if err != nil {
+		return s3response.PutObjectOutput{}, fmt.Errorf("remove replaced object attributes: %w", err)
 	}
 
 	// Before finalizing the object creation remove
@@ -4877,11 +4884,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, err
 		}
 		versionID = nullVersionId
-		// Clear any stale versionId sidecar attribute left from a previous
-		// versioned object at this path.  With xattr this is implicit (the
-		// new file carries only the attrs set on the tmpfile), but with
-		// path-based metadata the old attr persists until explicitly deleted.
-		_ = p.meta.DeleteAttribute(*po.Bucket, *po.Key, versionIdKey)
 	}
 
 	var sum string
@@ -5298,20 +5300,18 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 					return nil, fmt.Errorf("link tmp file: %w", err)
 				}
 
-				// With path-based metadata (sidecar) the live object's attrs are
-				// not replaced atomically.  The restored version may not have all
-				// the attrs that the deleted version had (e.g. a null version has
-				// no versionIdKey).  Clear attrs that belong to the deleted version
-				// before copying the restored version's attrs so that the restored
-				// version presents a clean state.
-				_ = p.meta.DeleteAttribute(bucket, object, versionIdKey)
-				_ = p.meta.DeleteAttribute(bucket, object, deleteMarkerKey)
-
 				attrs, err := p.meta.ListAttributes(versionPath, srcVersionId)
 				if err != nil {
 					return nil, fmt.Errorf("list object attributes: %w", err)
 				}
 
+				// the restored version has only its own attributes: the
+				// ones of the deleted version it doesn't have are removed
+				err = p.meta.ReplaceObject(bucket, object, attrs)
+				if err != nil {
+					return nil, fmt.Errorf("remove deleted version attributes: %w", err)
+				}
+
 				for _, attr := range attrs {
 					// the place of the null version only applies in
 					// the versioning directory

tests/integration/CompleteMultipartUpload.go

@@ -1745,6 +1745,113 @@ func CompleteMultipartUpload_default_content_type(s *S3Conf) error {
 	})
 }
 
+func CompleteMultipartUpload_overwrite_resets_attributes(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_overwrite_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		expires := time.Now().Add(time.Hour)
+
+		_, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:                  &bucket,
+			Key:                     &obj,
+			ContentType:             getPtr("text/plain"),
+			ContentEncoding:         getPtr("gzip"),
+			ContentDisposition:      getPtr("inline"),
+			ContentLanguage:         getPtr("en"),
+			CacheControl:            getPtr("no-cache"),
+			Expires:                 &expires,
+			WebsiteRedirectLocation: getPtr("/redirect"),
+			Metadata:                map[string]string{"foo": "bar"},
+			Tagging:                 getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		mp, err := createMp(s3client, bucket, obj)
+		if err != nil {
+			return err
+		}
+
+		parts, _, err := uploadParts(s3client, 20, 1, bucket, obj, *mp.UploadId)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: mp.UploadId,
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{
+						ETag:       parts[0].ETag,
+						PartNumber: parts[0].PartNumber,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the object holds none of the replaced object's attributes
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(out.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(out.ContentType))
+		}
+		for _, hdr := range []struct {
+			name  string
+			value *string
+		}{
+			{"Content-Encoding", out.ContentEncoding},
+			{"Content-Disposition", out.ContentDisposition},
+			{"Content-Language", out.ContentLanguage},
+			{"Cache-Control", out.CacheControl},
+			{"Expires", out.ExpiresString},
+			{"x-amz-website-redirect-location", out.WebsiteRedirectLocation},
+		} {
+			if hdr.value != nil {
+				return fmt.Errorf("expected nil %s, instead got %s", hdr.name, *hdr.value)
+			}
+		}
+		if len(out.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", out.Metadata)
+		}
+		if out.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *out.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		tagging, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(tagging.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", tagging.TagSet)
+		}
+
+		return nil
+	})
+}
+
 func CompleteMultipartUpload_success(s *S3Conf) error {
 	testName := "CompleteMultipartUpload_success"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {

tests/integration/CopyObject.go

@@ -847,6 +847,104 @@ func CopyObject_should_replace_meta_props(s *S3Conf) error {
 	})
 }
 
+func CopyObject_overwrite_resets_attributes(s *S3Conf) error {
+	testName := "CopyObject_overwrite_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		srcObj, dstObj := "source-object", "dest-object"
+		expires := time.Now().Add(time.Hour)
+
+		_, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:                  &bucket,
+			Key:                     &dstObj,
+			ContentType:             getPtr("text/plain"),
+			ContentEncoding:         getPtr("gzip"),
+			ContentDisposition:      getPtr("inline"),
+			ContentLanguage:         getPtr("en"),
+			CacheControl:            getPtr("no-cache"),
+			Expires:                 &expires,
+			WebsiteRedirectLocation: getPtr("/redirect"),
+			Metadata:                map[string]string{"foo": "bar"},
+			Tagging:                 getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		_, err = putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &srcObj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the source has no attributes to copy
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CopyObject(ctx, &s3.CopyObjectInput{
+			Bucket:     &bucket,
+			Key:        &dstObj,
+			CopySource: getPtr(bucket + "/" + srcObj),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the destination holds none of the replaced object's attributes
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &dstObj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(out.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(out.ContentType))
+		}
+		for _, hdr := range []struct {
+			name  string
+			value *string
+		}{
+			{"Content-Encoding", out.ContentEncoding},
+			{"Content-Disposition", out.ContentDisposition},
+			{"Content-Language", out.ContentLanguage},
+			{"Cache-Control", out.CacheControl},
+			{"Expires", out.ExpiresString},
+			{"x-amz-website-redirect-location", out.WebsiteRedirectLocation},
+		} {
+			if hdr.value != nil {
+				return fmt.Errorf("expected nil %s, instead got %s", hdr.name, *hdr.value)
+			}
+		}
+		if len(out.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", out.Metadata)
+		}
+		if out.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *out.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		tagging, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &dstObj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(tagging.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", tagging.TagSet)
+		}
+
+		return nil
+	})
+}
+
 func CopyObject_invalid_website_redirect_location(s *S3Conf) error {
 	testName := "CopyObject_invalid_website_redirect_location"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {

tests/integration/PutObject.go

@@ -1323,6 +1323,230 @@ func PutObject_default_content_type(s *S3Conf) error {
 	})
 }
 
+func PutObject_overwrite_resets_attributes(s *S3Conf) error {
+	testName := "PutObject_overwrite_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		expires := time.Now().Add(time.Hour)
+
+		_, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:                  &bucket,
+			Key:                     &obj,
+			ContentType:             getPtr("text/plain"),
+			ContentEncoding:         getPtr("gzip"),
+			ContentDisposition:      getPtr("inline"),
+			ContentLanguage:         getPtr("en"),
+			CacheControl:            getPtr("no-cache"),
+			Expires:                 &expires,
+			WebsiteRedirectLocation: getPtr("/redirect"),
+			Metadata:                map[string]string{"foo": "bar"},
+			Tagging:                 getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		_, err = putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the object holds none of the replaced object's attributes
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(out.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(out.ContentType))
+		}
+		for _, hdr := range []struct {
+			name  string
+			value *string
+		}{
+			{"Content-Encoding", out.ContentEncoding},
+			{"Content-Disposition", out.ContentDisposition},
+			{"Content-Language", out.ContentLanguage},
+			{"Cache-Control", out.CacheControl},
+			{"Expires", out.ExpiresString},
+			{"x-amz-website-redirect-location", out.WebsiteRedirectLocation},
+		} {
+			if hdr.value != nil {
+				return fmt.Errorf("expected nil %s, instead got %s", hdr.name, *hdr.value)
+			}
+		}
+		if len(out.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", out.Metadata)
+		}
+		if out.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *out.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		tagging, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(tagging.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", tagging.TagSet)
+		}
+
+		return nil
+	})
+}
+
+func PutObject_overwrite_multipart_object(s *S3Conf) error {
+	testName := "PutObject_overwrite_multipart_object"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		mp, err := createMp(s3client, bucket, obj)
+		if err != nil {
+			return err
+		}
+
+		parts, _, err := uploadParts(s3client, 100, 1, bucket, obj, *mp.UploadId)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: mp.UploadId,
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{
+						ETag:       parts[0].ETag,
+						PartNumber: parts[0].PartNumber,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		dataLen := int64(10)
+		_, err = putObjectWithData(dataLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the new object isn't a multipart upload: its only part is the
+		// whole object
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:     &bucket,
+			Key:        &obj,
+			PartNumber: getPtr(int32(1)),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.PartsCount != nil {
+			return fmt.Errorf("expected nil parts count, instead got %v", *out.PartsCount)
+		}
+		if out.ContentLength == nil {
+			return fmt.Errorf("expected non nil ContentLength")
+		}
+		if *out.ContentLength != dataLen {
+			return fmt.Errorf("expected the content length to be %v, instead got %v",
+				dataLen, *out.ContentLength)
+		}
+
+		return nil
+	})
+}
+
+func PutObject_dir_object_overwrite_resets_attributes(s *S3Conf) error {
+	testName := "PutObject_dir_object_overwrite_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-dir/"
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutObject(ctx, &s3.PutObjectInput{
+			Bucket:                  &bucket,
+			Key:                     &obj,
+			Tagging:                 getPtr("key1=val1&key2=val2"),
+			Metadata:                map[string]string{"foo": "bar"},
+			WebsiteRedirectLocation: getPtr("/some/redirect"),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.PutObject(ctx, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the object holds none of the replaced object's attributes
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(out.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", out.Metadata)
+		}
+		if out.WebsiteRedirectLocation != nil {
+			return fmt.Errorf("expected nil website-redirect-location, instead got %v",
+				*out.WebsiteRedirectLocation)
+		}
+		if out.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *out.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		tagging, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(tagging.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", tagging.TagSet)
+		}
+
+		return nil
+	})
+}
+
 func PutObject_invalid_credentials(s *S3Conf) error {
 	testName := "PutObject_invalid_credentials"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {

tests/integration/group-tests.go

@@ -206,6 +206,9 @@ func TestPutObject(ts *TestState) {
 	}
 	ts.Run(PutObject_success)
 	ts.Run(PutObject_default_content_type)
+	ts.Run(PutObject_overwrite_resets_attributes)
+	ts.Run(PutObject_overwrite_multipart_object)
+	ts.Run(PutObject_dir_object_overwrite_resets_attributes)
 	if !ts.conf.versioningEnabled && !ts.conf.windowsTests {
 		ts.Run(PutObject_racey_success)
 	}
@@ -390,6 +393,7 @@ func TestCopyObject(ts *TestState) {
 	ts.Run(CopyObject_should_copy_meta_props)
 	ts.Run(CopyObject_should_not_copy_website_redirect_without_user_metadata)
 	ts.Run(CopyObject_should_replace_meta_props)
+	ts.Run(CopyObject_overwrite_resets_attributes)
 	ts.Run(CopyObject_invalid_website_redirect_location)
 	ts.Run(CopyObject_default_content_type_with_replace_metadata)
 	ts.Run(CopyObject_missing_bucket_lock)
@@ -561,6 +565,7 @@ func TestCompleteMultipartUpload(ts *TestState) {
 	ts.Run(CompleteMultipartUpload_incorrect_part_number)
 	ts.Run(CompleteMultipartUpload_invalid_part_number)
 	ts.Run(CompleteMultipartUpload_default_content_type)
+	ts.Run(CompleteMultipartUpload_overwrite_resets_attributes)
 	ts.Run(CompleteMultipartUpload_invalid_ETag)
 	ts.Run(CompleteMultipartUpload_small_upload_size)
 	ts.Run(CompleteMultipartUpload_empty_parts)
@@ -1988,6 +1993,9 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_PutObject_overwrite_null_versionId_obj)
 	ts.Run(Versioning_PutObject_success)
 	ts.Run(Versioning_PutObject_dir_object_new_version_resets_attributes)
+	ts.Run(Versioning_PutObject_new_version_resets_attributes)
+	ts.Run(Versioning_PutObject_over_delete_marker_resets_attributes)
+	ts.Run(Versioning_PutObject_suspended_null_version_resets_attributes)
 	// CopyObject action
 	ts.Run(Versioning_CopyObject_invalid_versionId)
 	ts.Run(Versioning_CopyObject_encoded_versionid_separator_invalid_versionId)
@@ -2041,6 +2049,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_DeleteObject_dir_object_latest_version)
 	ts.Run(Versioning_DeleteObject_latest_version_with_null_version)
 	ts.Run(Versioning_DeleteObject_latest_version_null_version_order)
+	ts.Run(Versioning_DeleteObject_promoted_version_attributes)
 	ts.Run(Versioning_DeleteObject_non_existing_object)
 	ts.Run(Versioning_DeleteObject_implicit_dir)
 	ts.Run(Versioning_DeleteObject_trailing_slash_counterpart)
@@ -2113,6 +2122,8 @@ func TestVersioning(ts *TestState) {
 	}
 	ts.Run(Versioning_WORM_trailing_slash_counterpart)
 	ts.Run(Versioning_WORM_null_version_locked_with_legal_hold)
+	ts.Run(Versioning_WORM_PutObject_new_version_lock_settings)
+	ts.Run(Versioning_WORM_DeleteObject_promoted_version_lock_settings)
 	// Concurrent requests
 	// Versioninig_concurrent_upload_object
 	ts.Run(Versioning_AccessControl_GetObjectVersion)
@@ -2968,6 +2979,9 @@ func GetIntTests() IntTests {
 		"PutObject_tagging":                                                                PutObject_tagging,
 		"PutObject_success":                                                                PutObject_success,
 		"PutObject_default_content_type":                                                   PutObject_default_content_type,
+		"PutObject_overwrite_resets_attributes":                                            PutObject_overwrite_resets_attributes,
+		"PutObject_overwrite_multipart_object":                                             PutObject_overwrite_multipart_object,
+		"PutObject_dir_object_overwrite_resets_attributes":                                 PutObject_dir_object_overwrite_resets_attributes,
 		"PutObject_invalid_object_names":                                                   PutObject_invalid_object_names,
 		"PutObject_object_acl_not_supported":                                               PutObject_object_acl_not_supported,
 		"PutObject_false_negative_object_names":                                            PutObject_false_negative_object_names,
@@ -3114,6 +3128,7 @@ func GetIntTests() IntTests {
 		"CopyObject_should_copy_meta_props":                                                CopyObject_should_copy_meta_props,
 		"CopyObject_should_not_copy_website_redirect_without_user_metadata":                CopyObject_should_not_copy_website_redirect_without_user_metadata,
 		"CopyObject_should_replace_meta_props":                                             CopyObject_should_replace_meta_props,
+		"CopyObject_overwrite_resets_attributes":                                           CopyObject_overwrite_resets_attributes,
 		"CopyObject_invalid_website_redirect_location":                                     CopyObject_invalid_website_redirect_location,
 		"CopyObject_default_content_type_with_replace_metadata":                            CopyObject_default_content_type_with_replace_metadata,
 		"CopyObject_missing_bucket_lock":                                                   CopyObject_missing_bucket_lock,
@@ -3234,6 +3249,7 @@ func GetIntTests() IntTests {
 		"CompletedMultipartUpload_non_existing_bucket":                                     CompletedMultipartUpload_non_existing_bucket,
 		"CompleteMultipartUpload_invalid_part_number":                                      CompleteMultipartUpload_invalid_part_number,
 		"CompleteMultipartUpload_default_content_type":                                     CompleteMultipartUpload_default_content_type,
+		"CompleteMultipartUpload_overwrite_resets_attributes":                              CompleteMultipartUpload_overwrite_resets_attributes,
 		"CompleteMultipartUpload_invalid_ETag":                                             CompleteMultipartUpload_invalid_ETag,
 		"CompleteMultipartUpload_small_upload_size":                                        CompleteMultipartUpload_small_upload_size,
 		"CompleteMultipartUpload_empty_parts":                                              CompleteMultipartUpload_empty_parts,
@@ -3584,6 +3600,9 @@ func GetIntTests() IntTests {
 		"Versioning_PutObject_overwrite_null_versionId_obj":                                Versioning_PutObject_overwrite_null_versionId_obj,
 		"Versioning_PutObject_success":                                                     Versioning_PutObject_success,
 		"Versioning_PutObject_dir_object_new_version_resets_attributes":                    Versioning_PutObject_dir_object_new_version_resets_attributes,
+		"Versioning_PutObject_new_version_resets_attributes":                               Versioning_PutObject_new_version_resets_attributes,
+		"Versioning_PutObject_over_delete_marker_resets_attributes":                        Versioning_PutObject_over_delete_marker_resets_attributes,
+		"Versioning_PutObject_suspended_null_version_resets_attributes":                    Versioning_PutObject_suspended_null_version_resets_attributes,
 		"Versioning_CopyObject_invalid_versionId":                                          Versioning_CopyObject_invalid_versionId,
 		"Versioning_CopyObject_encoded_versionid_separator_invalid_versionId":              Versioning_CopyObject_encoded_versionid_separator_invalid_versionId,
 		"Versioning_CopyObject_success":                                                    Versioning_CopyObject_success,
@@ -3626,6 +3645,7 @@ func GetIntTests() IntTests {
 		"Versioning_DeleteObject_delete_object_version":                                    Versioning_DeleteObject_delete_object_version,
 		"Versioning_DeleteObject_latest_version_with_null_version":                         Versioning_DeleteObject_latest_version_with_null_version,
 		"Versioning_DeleteObject_latest_version_null_version_order":                        Versioning_DeleteObject_latest_version_null_version_order,
+		"Versioning_DeleteObject_promoted_version_attributes":                              Versioning_DeleteObject_promoted_version_attributes,
 		"Versioning_DeleteObject_dir_object_latest_version":                                Versioning_DeleteObject_dir_object_latest_version,
 		"Versioning_DeleteObject_non_existing_object":                                      Versioning_DeleteObject_non_existing_object,
 		"Versioning_DeleteObject_implicit_dir":                                             Versioning_DeleteObject_implicit_dir,
@@ -3689,6 +3709,8 @@ func GetIntTests() IntTests {
 		"Versioning_WORM_remove_delete_marker_under_bucket_default_retention":              Versioning_WORM_remove_delete_marker_under_bucket_default_retention,
 		"Versioning_WORM_trailing_slash_counterpart":                                       Versioning_WORM_trailing_slash_counterpart,
 		"Versioning_WORM_null_version_locked_with_legal_hold":                              Versioning_WORM_null_version_locked_with_legal_hold,
+		"Versioning_WORM_PutObject_new_version_lock_settings":                              Versioning_WORM_PutObject_new_version_lock_settings,
+		"Versioning_WORM_DeleteObject_promoted_version_lock_settings":                      Versioning_WORM_DeleteObject_promoted_version_lock_settings,
 		"Versioning_AccessControl_GetObjectVersion":                                        Versioning_AccessControl_GetObjectVersion,
 		"Versioning_AccessControl_HeadObjectVersion":                                       Versioning_AccessControl_HeadObjectVersion,
 		"Versioning_AccessControl_object_tagging_policy":                                   Versioning_AccessControl_object_tagging_policy,

tests/integration/versioning.go

@@ -415,6 +415,283 @@ func Versioning_PutObject_dir_object_new_version_resets_attributes(s *S3Conf) er
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_PutObject_new_version_resets_attributes(s *S3Conf) error {
+	testName := "Versioning_PutObject_new_version_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		meta := map[string]string{"foo": "bar"}
+		tags := []types.Tag{{Key: getPtr("key"), Value: getPtr("val")}}
+
+		first, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:          &bucket,
+			Key:             &obj,
+			ContentType:     getPtr("text/plain"),
+			ContentEncoding: getPtr("gzip"),
+			CacheControl:    getPtr("no-cache"),
+			Metadata:        meta,
+			Tagging:         getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		second, err := putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the current version holds none of the first version attributes
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != getString(second.res.VersionId) {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				getString(second.res.VersionId), getString(cur.VersionId))
+		}
+		if getString(cur.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(cur.ContentType))
+		}
+		if cur.ContentEncoding != nil {
+			return fmt.Errorf("expected nil Content-Encoding, instead got %v", *cur.ContentEncoding)
+		}
+		if cur.CacheControl != nil {
+			return fmt.Errorf("expected nil Cache-Control, instead got %v", *cur.CacheControl)
+		}
+		if len(cur.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", cur.Metadata)
+		}
+		if cur.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *cur.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		curTags, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(curTags.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", curTags.TagSet)
+		}
+
+		// the first version keeps its attributes
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		old, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: first.res.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(old.ContentType) != "text/plain" {
+			return fmt.Errorf("expected the content-type to be text/plain, instead got %v",
+				getString(old.ContentType))
+		}
+		if getString(old.ContentEncoding) != "gzip" {
+			return fmt.Errorf("expected the content-encoding to be gzip, instead got %v",
+				getString(old.ContentEncoding))
+		}
+		if !areMapsSame(meta, old.Metadata) {
+			return fmt.Errorf("expected the metadata to be %v, instead got %v",
+				meta, old.Metadata)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		oldTags, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: first.res.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !areTagsSame(tags, oldTags.TagSet) {
+			return fmt.Errorf("expected the tag set to be %v, instead got %v",
+				tags, oldTags.TagSet)
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+func Versioning_PutObject_over_delete_marker_resets_attributes(s *S3Conf) error {
+	testName := "Versioning_PutObject_over_delete_marker_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+
+		_, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:          &bucket,
+			Key:             &obj,
+			ContentType:     getPtr("text/plain"),
+			ContentEncoding: getPtr("gzip"),
+			CacheControl:    getPtr("no-cache"),
+			Metadata:        map[string]string{"foo": "bar"},
+			Tagging:         getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		_, err = createDeleteMarker(s3client, bucket, obj)
+		if err != nil {
+			return err
+		}
+
+		res, err := putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the new version holds none of the attributes of the version
+		// under the delete marker
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != getString(res.res.VersionId) {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				getString(res.res.VersionId), getString(cur.VersionId))
+		}
+		if getString(cur.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(cur.ContentType))
+		}
+		if cur.ContentEncoding != nil {
+			return fmt.Errorf("expected nil Content-Encoding, instead got %v", *cur.ContentEncoding)
+		}
+		if cur.CacheControl != nil {
+			return fmt.Errorf("expected nil Cache-Control, instead got %v", *cur.CacheControl)
+		}
+		if len(cur.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", cur.Metadata)
+		}
+		if cur.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *cur.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		curTags, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(curTags.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", curTags.TagSet)
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+func Versioning_PutObject_suspended_null_version_resets_attributes(s *S3Conf) error {
+	testName := "Versioning_PutObject_suspended_null_version_resets_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+
+		_, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:          &bucket,
+			Key:             &obj,
+			ContentType:     getPtr("text/plain"),
+			ContentEncoding: getPtr("gzip"),
+			CacheControl:    getPtr("no-cache"),
+			Metadata:        map[string]string{"foo": "bar"},
+			Tagging:         getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the null version is replaced by the new null version
+		_, err = putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != nullVersionId {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				nullVersionId, getString(cur.VersionId))
+		}
+		if getString(cur.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(cur.ContentType))
+		}
+		if cur.ContentEncoding != nil {
+			return fmt.Errorf("expected nil Content-Encoding, instead got %v", *cur.ContentEncoding)
+		}
+		if cur.CacheControl != nil {
+			return fmt.Errorf("expected nil Cache-Control, instead got %v", *cur.CacheControl)
+		}
+		if len(cur.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", cur.Metadata)
+		}
+		if cur.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *cur.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		curTags, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(curTags.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", curTags.TagSet)
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusSuspended))
+}
+
 func Versioning_CopyObject_invalid_versionId(s *S3Conf) error {
 	testName := "Versioning_CopyObject_invalid_versionId"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -2431,6 +2708,116 @@ func Versioning_DeleteObject_latest_version_null_version_order(s *S3Conf) error
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_DeleteObject_promoted_version_attributes(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_promoted_version_attributes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		expires := time.Now().Add(time.Hour)
+
+		dataLen := int64(10)
+		first, err := putObjectWithData(dataLen, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		second, err := putObjectWithData(20, &s3.PutObjectInput{
+			Bucket:                  &bucket,
+			Key:                     &obj,
+			ContentType:             getPtr("text/plain"),
+			ContentEncoding:         getPtr("gzip"),
+			ContentDisposition:      getPtr("inline"),
+			ContentLanguage:         getPtr("en"),
+			CacheControl:            getPtr("no-cache"),
+			Expires:                 &expires,
+			WebsiteRedirectLocation: getPtr("/redirect"),
+			Metadata:                map[string]string{"foo": "bar"},
+			Tagging:                 getPtr("key=val"),
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// deleting the latest version makes the first version current again
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: second.res.VersionId,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the first version holds none of the deleted version attributes
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != getString(first.res.VersionId) {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				getString(first.res.VersionId), getString(cur.VersionId))
+		}
+		if cur.ContentLength == nil {
+			return fmt.Errorf("expected non nil ContentLength")
+		}
+		if *cur.ContentLength != dataLen {
+			return fmt.Errorf("expected the content length to be %v, instead got %v",
+				dataLen, *cur.ContentLength)
+		}
+		if getString(cur.ContentType) != defaultContentType {
+			return fmt.Errorf("expected default %s Content-Type, instead got %s",
+				defaultContentType, getString(cur.ContentType))
+		}
+		for _, hdr := range []struct {
+			name  string
+			value *string
+		}{
+			{"Content-Encoding", cur.ContentEncoding},
+			{"Content-Disposition", cur.ContentDisposition},
+			{"Content-Language", cur.ContentLanguage},
+			{"Cache-Control", cur.CacheControl},
+			{"Expires", cur.ExpiresString},
+			{"x-amz-website-redirect-location", cur.WebsiteRedirectLocation},
+		} {
+			if hdr.value != nil {
+				return fmt.Errorf("expected nil %s, instead got %s", hdr.name, *hdr.value)
+			}
+		}
+		if len(cur.Metadata) != 0 {
+			return fmt.Errorf("expected empty metadata, instead got %v", cur.Metadata)
+		}
+		if cur.TagCount != nil {
+			return fmt.Errorf("expected nil tag count, instead got %v", *cur.TagCount)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		curTags, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(curTags.TagSet) != 0 {
+			return fmt.Errorf("expected empty tag set, instead got %v", curTags.TagSet)
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func Versioning_DeleteObject_non_existing_object(s *S3Conf) error {
 	testName := "Versioning_DeleteObject_non_existing_object"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5731,6 +6118,152 @@ func Versioning_WORM_null_version_locked_with_legal_hold(s *S3Conf) error {
 	})
 }
 
+func Versioning_WORM_PutObject_new_version_lock_settings(s *S3Conf) error {
+	testName := "Versioning_WORM_PutObject_new_version_lock_settings"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		rDate := time.Now().Add(time.Hour * 48)
+
+		first, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket:                    &bucket,
+			Key:                       &obj,
+			ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+			ObjectLockMode:            types.ObjectLockModeGovernance,
+			ObjectLockRetainUntilDate: &rDate,
+		}, s3client, withPutObjectChecksumAlgo(types.ChecksumAlgorithmCrc32))
+		if err != nil {
+			return err
+		}
+
+		second, err := putObjectWithData(20, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		// the new version isn't locked by the first version lock settings
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != getString(second.res.VersionId) {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				getString(second.res.VersionId), getString(cur.VersionId))
+		}
+		if cur.ObjectLockLegalHoldStatus != "" {
+			return fmt.Errorf("expected empty object legal hold status, instead got %q",
+				cur.ObjectLockLegalHoldStatus)
+		}
+		if cur.ObjectLockMode != "" {
+			return fmt.Errorf("expected empty object lock mode, instead got %q", cur.ObjectLockMode)
+		}
+		if cur.ObjectLockRetainUntilDate != nil {
+			return fmt.Errorf("expected nil object lock retain until date, instead got %v",
+				*cur.ObjectLockRetainUntilDate)
+		}
+
+		return cleanupLockedObjects(s3client, bucket, []objToDelete{
+			{
+				key:             obj,
+				versionId:       getString(first.res.VersionId),
+				removeLegalHold: true,
+			},
+		})
+	}, withLock())
+}
+
+func Versioning_WORM_DeleteObject_promoted_version_lock_settings(s *S3Conf) error {
+	testName := "Versioning_WORM_DeleteObject_promoted_version_lock_settings"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj := "my-obj"
+		rDate := time.Now().Add(time.Hour * 48)
+
+		first, err := putObjectWithData(10, &s3.PutObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		}, s3client)
+		if err != nil {
+			return err
+		}
+
+		second, err := putObjectWithData(20, &s3.PutObjectInput{
+			Bucket:                    &bucket,
+			Key:                       &obj,
+			ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+			ObjectLockMode:            types.ObjectLockModeGovernance,
+			ObjectLockRetainUntilDate: &rDate,
+		}, s3client, withPutObjectChecksumAlgo(types.ChecksumAlgorithmCrc32))
+		if err != nil {
+			return err
+		}
+
+		// release the latest version and delete it, which makes the first
+		// version current again
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.PutObjectLegalHold(ctx, &s3.PutObjectLegalHoldInput{
+			Bucket:    &bucket,
+			Key:       &obj,
+			VersionId: second.res.VersionId,
+			LegalHold: &types.ObjectLockLegalHold{
+				Status: types.ObjectLockLegalHoldStatusOff,
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket:                    &bucket,
+			Key:                       &obj,
+			VersionId:                 second.res.VersionId,
+			BypassGovernanceRetention: getBoolPtr(true),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the first version holds none of the deleted version lock settings
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		cur, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &obj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(cur.VersionId) != getString(first.res.VersionId) {
+			return fmt.Errorf("expected the current versionId to be %v, instead got %v",
+				getString(first.res.VersionId), getString(cur.VersionId))
+		}
+		if cur.ObjectLockLegalHoldStatus != "" {
+			return fmt.Errorf("expected empty object legal hold status, instead got %q",
+				cur.ObjectLockLegalHoldStatus)
+		}
+		if cur.ObjectLockMode != "" {
+			return fmt.Errorf("expected empty object lock mode, instead got %q", cur.ObjectLockMode)
+		}
+		if cur.ObjectLockRetainUntilDate != nil {
+			return fmt.Errorf("expected nil object lock retain until date, instead got %v",
+				*cur.ObjectLockRetainUntilDate)
+		}
+
+		return nil
+	}, withLock())
+}
+
 func Versioning_AccessControl_GetObjectVersion(s *S3Conf) error {
 	testName := "Versioning_AccessControl_GetObjectVersion"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {