Merge pull request #2413 from versity/sis/directory-objects-versioning

feat: add versioning support for directory objects in posix

auth/object_lock.go

@@ -495,7 +495,9 @@ func (s objectLockState) checkObject(ctx context.Context, be backend.Backend, ia
 
 	checkRetention := true
 	retentionData, err := be.GetObjectRetention(ctx, bucket, key, versionId)
-	if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) {
+	// an object or version that doesn't exist has nothing to protect
+	if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) ||
+		errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchVersion)) {
 		return nil
 	}
 	// the object is a delete marker, if a `MethodNotAllowed` error is returned
@@ -538,7 +540,8 @@ func (s objectLockState) checkObject(ctx context.Context, be backend.Backend, ia
 
 	status, err := be.GetObjectLegalHold(ctx, bucket, key, versionId)
 	if err != nil {
-		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) {
+		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) ||
+			errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchVersion)) {
 			return nil
 		}
 		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchObjectLockConfiguration)) {

backend/posix/objlock.go

@@ -49,6 +49,11 @@ import (
 // file takes its place, which is unsafe to detect reliably on NFS due to
 // attribute caching.
 //
+// The process-local slot is picked from the bucket hash and the shard, so all
+// requests for one lock file take the same slot (fcntl locks don't exclude
+// within a process), while the same key in different buckets usually takes
+// different slots and doesn't wait on an unrelated lock.
+//
 // The lock is held only for the commit phase (condition re-check, metadata
 // stores, final link/rename) — request bodies are staged to a temp file
 // before the lock is taken. The OS releases advisory locks automatically when
@@ -131,15 +136,22 @@ func objLockShard(object string) uint8 {
 	return sum[0]
 }
 
+// objLockSlot returns the process-local slot index for the lock file of the
+// shard in the bucket with bucketHash.
+func objLockSlot(bucketHash [sha256.Size]byte, shard uint8) uint8 {
+	return bucketHash[0] ^ shard
+}
+
 // lockObjectPublish acquires the publish lock for bucket/object. It returns a
 // release function that must be called (typically deferred) once the new
 // object state is visible. All code paths that create or replace an object at
 // its final key must hold this lock across condition evaluation and
 // publication.
 func (p *Posix) lockObjectPublish(ctx context.Context, bucket, object string) (func(), error) {
+	bucketHash := sha256.Sum256([]byte(bucket))
 	shard := objLockShard(object)
 
-	slot := p.objLockSlots[shard]
+	slot := p.objLockSlots[objLockSlot(bucketHash, shard)]
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
@@ -154,7 +166,7 @@ func (p *Posix) lockObjectPublish(ctx context.Context, bucket, object string) (f
 		return releaseLocal, nil
 	}
 
-	f, err := p.openObjLockFile(bucket, shard)
+	f, err := p.openObjLockFile(bucketHash, shard)
 	if err != nil {
 		releaseLocal()
 		return nil, err
@@ -188,9 +200,8 @@ func newObjLockSlots() [objLockShards]chan struct{} {
 }
 
 // openObjLockFile opens (creating as needed) the lock file for the shard in
-// the given bucket.
-func (p *Posix) openObjLockFile(bucket string, shard uint8) (*os.File, error) {
-	bucketHash := sha256.Sum256([]byte(bucket))
+// the bucket with bucketHash.
+func (p *Posix) openObjLockFile(bucketHash [sha256.Size]byte, shard uint8) (*os.File, error) {
 	lockDir := filepath.Join(p.rootdir, objLockDir, fmt.Sprintf("%x", bucketHash))
 	name := filepath.Join(lockDir, fmt.Sprintf("%02x", shard))
 

backend/posix/posix_conditional_put_test.go

@@ -17,6 +17,7 @@ package posix
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
 	"errors"
 	"fmt"
 	"io"
@@ -85,9 +86,9 @@ func TestObjectPublishLockHonorsContextWhileWaiting(t *testing.T) {
 	bucket := "testbucket"
 	createTestBucket(t, p, bucket)
 
-	shard := objLockShard("cancel-wait")
-	<-p.objLockSlots[shard]
-	defer func() { p.objLockSlots[shard] <- struct{}{} }()
+	slot := objLockSlot(sha256.Sum256([]byte(bucket)), objLockShard("cancel-wait"))
+	<-p.objLockSlots[slot]
+	defer func() { p.objLockSlots[slot] <- struct{}{} }()
 	if _, err := os.Stat(p.ObjectPath(bucket, objLockDir)); !errors.Is(err, fs.ErrNotExist) {
 		t.Fatalf("bucket contains publish lock directory: %v", err)
 	}
@@ -134,6 +135,25 @@ func TestObjectPublishLockHonorsCancellationAfterSlotAcquired(t *testing.T) {
 	}
 }
 
+func TestObjectPublishLockDoesNotBlockOtherBuckets(t *testing.T) {
+	p := newTestPosix(t, metaModes(t)["xattr"])
+
+	unlock, err := p.lockObjectPublish(context.Background(), "bucket-a", "my-obj")
+	if err != nil {
+		t.Fatalf("lock object publish: %v", err)
+	}
+	defer unlock()
+
+	// the same key in another bucket has its own lock file
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	unlockOther, err := p.lockObjectPublish(ctx, "bucket-b", "my-obj")
+	if err != nil {
+		t.Fatalf("lock object publish in another bucket: %v", err)
+	}
+	unlockOther()
+}
+
 func TestObjectPublishLockModes(t *testing.T) {
 	tests := []struct {
 		name  string

backend/posix/posix_io_helpers.go

@@ -16,6 +16,7 @@ package posix
 
 import (
 	"bufio"
+	"errors"
 	"io"
 	"log"
 	"sync"
@@ -45,6 +46,24 @@ func (b *bufferedReadCloser) Close() error {
 	return b.c.Close()
 }
 
+// closeOnEOFReader closes c once r is drained, for readers whose source has
+// to be released before the caller is done with the reader.
+type closeOnEOFReader struct {
+	r      io.Reader
+	c      io.Closer
+	closed bool
+}
+
+func (e *closeOnEOFReader) Read(p []byte) (int, error) {
+	n, err := e.r.Read(p)
+	if errors.Is(err, io.EOF) && !e.closed {
+		e.closed = true
+		e.c.Close()
+	}
+
+	return n, err
+}
+
 var odirectUnsupportedWarnByOp sync.Map
 
 func warnODirectUnsupportedOnce(op string, err error) {

backend/posix/posix_io_helpers_test.go

@@ -0,0 +1,79 @@
+// 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 posix
+
+import (
+	"bytes"
+	"io"
+	"strings"
+	"testing"
+)
+
+type countingCloser struct {
+	count int
+}
+
+func (c *countingCloser) Close() error {
+	c.count++
+	return nil
+}
+
+func TestCloseOnEOFReader(t *testing.T) {
+	c := &countingCloser{}
+	r := &closeOnEOFReader{r: strings.NewReader("hello"), c: c}
+
+	buf := make([]byte, 2)
+	n, err := r.Read(buf)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if n != 2 {
+		t.Fatalf("expected 2 bytes, got %v", n)
+	}
+	if c.count != 0 {
+		t.Fatalf("expected the source to stay open before EOF, closed %v times", c.count)
+	}
+
+	data, err := io.ReadAll(r)
+	if err != nil {
+		t.Fatalf("read all: %v", err)
+	}
+	if !bytes.Equal(data, []byte("llo")) {
+		t.Fatalf("expected the remaining data to be llo, got %s", data)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once at EOF, closed %v times", c.count)
+	}
+
+	// reads past EOF don't close the source again
+	if _, err := r.Read(buf); err != io.EOF {
+		t.Fatalf("expected io.EOF, got %v", err)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once, closed %v times", c.count)
+	}
+}
+
+func TestCloseOnEOFReaderEmptySource(t *testing.T) {
+	c := &countingCloser{}
+	r := &closeOnEOFReader{r: strings.NewReader(""), c: c}
+
+	if _, err := io.ReadAll(r); err != nil {
+		t.Fatalf("read all: %v", err)
+	}
+	if c.count != 1 {
+		t.Fatalf("expected the source to be closed once at EOF, closed %v times", c.count)
+	}
+}

backend/walk.go

@@ -177,11 +177,18 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			return fs.SkipDir
 		}
 
+		// A directory is compared to the markers as the key of its
+		// directory object, which has a trailing slash.
+		key := path
+		if d.IsDir() {
+			key += "/"
+		}
+
 		if !pastMarker {
-			if path == keyMarker {
+			if key == keyMarker {
 				pastMarker = true
 			}
-			if path < keyMarker {
+			if key < keyMarker {
 				return nil
 			}
 		}
@@ -195,28 +202,32 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			// building to match. So only skip if path isn't a prefix of prefix
 			// and prefix isn't a prefix of path.
 			if prefix != "" &&
-				!strings.HasPrefix(path+"/", prefix) &&
-				!strings.HasPrefix(prefix, path+"/") {
+				!strings.HasPrefix(key, prefix) &&
+				!strings.HasPrefix(prefix, key) {
 				return fs.SkipDir
 			}
 
 			// Don't recurse into subdirectories when listing with delimiter.
+			// The walk also enters the directories that sort before the key
+			// marker, so the common prefix is the part of the key up to the
+			// first delimiter after the prefix.
 			if delimiter == "/" &&
-				prefix != path+"/" &&
-				strings.HasPrefix(path+"/", prefix) {
-				cpmap.Add(path + "/")
+				prefix != key &&
+				strings.HasPrefix(key, prefix) {
+				before, _, _ := strings.Cut(strings.TrimPrefix(key, prefix), delimiter)
+				cpmap.Add(prefix + before + delimiter)
 				return fs.SkipDir
 			}
 
 			// Skip ancestor directories of the specified prefix; only process
 			// the directory that exactly matches the prefix.
-			// At this point we know strings.HasPrefix(prefix, path+"/") holds
+			// At this point we know strings.HasPrefix(prefix, key) holds
 			// (i.e. path is an ancestor of the prefix directory). Skip it
 			// unless it is the exact prefix directory.
 			// Note: WalkVersions always walks from "." (unlike Walk, which
 			// narrows the root) because versioning marker semantics require
 			// visiting all entries in order, so this guard is needed instead.
-			if prefix != "" && strings.HasPrefix(prefix, path+"/") && path+"/" != prefix {
+			if prefix != "" && strings.HasPrefix(prefix, key) && key != prefix {
 				return nil
 			}
 
@@ -231,7 +242,7 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 			delMarkers = append(delMarkers, res.DelMarkers...)
 			if res.Truncated {
 				truncated = true
-				nextMarker = path
+				nextMarker = key
 				nextVersionIdMarker = res.NextVersionIdMarker
 				return fs.SkipAll
 			}

s3err/s3err.go

@@ -131,6 +131,7 @@ const (
 	ErrMissingDateHeader
 	ErrGetUploadsWithKey
 	ErrVersionsWithKey
+	ErrCopySourceDeleteMarker
 	ErrInvalidRequest
 	ErrAuthNotSetup
 	ErrNotImplemented
@@ -453,6 +454,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "There is no such thing as the ?versions sub-resource for a key",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrCopySourceDeleteMarker: {
+		Code:           "InvalidRequest",
+		Description:    "The source of a copy request may not specifically refer to a delete marker by version id.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrInvalidRequest: {
 		Code:           "InvalidRequest",
 		Description:    "Invalid Request.",

tests/integration/ListObjectVersions.go

@@ -26,7 +26,7 @@ import (
 func ListObjectVersions_VD_success(s *S3Conf) error {
 	testName := "ListObjectVersions_VD_success"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		versions := []types.ObjectVersion{}
+		versions, dirVersions := []types.ObjectVersion{}, []types.ObjectVersion{}
 		for i := range 5 {
 			dLgth := int64(i * 100)
 			key := fmt.Sprintf("my-obj-%v", i)
@@ -46,8 +46,29 @@ func ListObjectVersions_VD_success(s *S3Conf) error {
 				VersionId:    getPtr("null"),
 				StorageClass: types.ObjectVersionStorageClassStandard,
 			})
+
+			dir := fmt.Sprintf("my-dir-%v/", i)
+			out, err = putObjectWithData(0, &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &dir,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+
+			dirVersions = append(dirVersions, types.ObjectVersion{
+				ETag:         out.res.ETag,
+				IsLatest:     getBoolPtr(true),
+				Key:          &dir,
+				Size:         getPtr(int64(0)),
+				VersionId:    getPtr("null"),
+				StorageClass: types.ObjectVersionStorageClassStandard,
+			})
 		}
 
+		// the directory objects are listed before the regular objects
+		versions = append(dirVersions, versions...)
+
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
 			Bucket: &bucket,
@@ -97,11 +118,68 @@ func ListObjectVersions_negative_max_keys(s *S3Conf) error {
 func ListObjectVersions_list_single_object_versions(s *S3Conf) error {
 	testName := "ListObjectVersions_list_single_object_versions"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		object := "my-obj"
-		versions, err := createObjVersions(s3client, bucket, object, 5)
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(object string) error {
+			versions, err := createObjVersions(s3client, bucket, object, 5)
+			if err != nil {
+				return err
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket: &bucket,
+				Prefix: &object,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if !compareVersions(versions, out.Versions) {
+				return fmt.Errorf("expected the resulting versions to be %v, instead got %v",
+					versions, out.Versions)
+			}
+
+			return nil
+		})
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+func ListObjectVersions_list_multiple_object_versions(s *S3Conf) error {
+	testName := "ListObjectVersions_list_multiple_object_versions"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		obj1, obj2, obj3 := "foo", "bar", "baz"
+		// "qux-quux" sorts before the directory object "qux/", which
+		// sorts before the object under it
+		dir, beforeDir, inDir := "qux/", "qux-quux", "qux/quux"
+
+		obj1Versions, err := createObjVersions(s3client, bucket, obj1, 4)
 		if err != nil {
 			return err
 		}
+		obj2Versions, err := createObjVersions(s3client, bucket, obj2, 3)
+		if err != nil {
+			return err
+		}
+		obj3Versions, err := createObjVersions(s3client, bucket, obj3, 5)
+		if err != nil {
+			return err
+		}
+		// "qux/" is put over the existing parent directory of "qux/quux"
+		inDirVersions, err := createObjVersions(s3client, bucket, inDir, 2)
+		if err != nil {
+			return err
+		}
+		dirVersions, err := createObjVersions(s3client, bucket, dir, 3)
+		if err != nil {
+			return err
+		}
+		beforeDirVersions, err := createObjVersions(s3client, bucket, beforeDir, 2)
+		if err != nil {
+			return err
+		}
+
+		versions := append(append(obj2Versions, obj3Versions...), obj1Versions...)
+		versions = append(append(append(versions, beforeDirVersions...), dirVersions...), inDirVersions...)
 
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
@@ -121,38 +199,100 @@ func ListObjectVersions_list_single_object_versions(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
-func ListObjectVersions_list_multiple_object_versions(s *S3Conf) error {
-	testName := "ListObjectVersions_list_multiple_object_versions"
+func ListObjectVersions_dir_object_with_children(s *S3Conf) error {
+	testName := "ListObjectVersions_dir_object_with_children"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj1, obj2, obj3 := "foo", "bar", "baz"
+		obj, dir, child, nestedObj := "my-obj", "my-dir/", "my-dir/child", "my-dir/sub/obj"
 
-		obj1Versions, err := createObjVersions(s3client, bucket, obj1, 4)
+		objVersions, err := createObjVersions(s3client, bucket, obj, 2)
 		if err != nil {
 			return err
 		}
-		obj2Versions, err := createObjVersions(s3client, bucket, obj2, 3)
+		dirVersions, err := createObjVersions(s3client, bucket, dir, 3)
 		if err != nil {
 			return err
 		}
-		obj3Versions, err := createObjVersions(s3client, bucket, obj3, 5)
+		childVersions, err := createObjVersions(s3client, bucket, child, 2)
+		if err != nil {
+			return err
+		}
+		nestedVersions, err := createObjVersions(s3client, bucket, nestedObj, 1)
 		if err != nil {
 			return err
 		}
 
-		versions := append(append(obj2Versions, obj3Versions...), obj1Versions...)
-
+		// the objects under "my-dir/" stay listed after it's deleted
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+		res, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
 			Bucket: &bucket,
+			Key:    &dir,
 		})
 		cancel()
 		if err != nil {
 			return err
 		}
 
-		if !compareVersions(versions, out.Versions) {
-			return fmt.Errorf("expected the resulting versions to be %v, instead got %v",
-				versions, out.Versions)
+		dirVersions[0].IsLatest = getBoolPtr(false)
+		delMarkers := []types.DeleteMarkerEntry{
+			{
+				IsLatest:  getBoolPtr(true),
+				Key:       &dir,
+				VersionId: res.VersionId,
+			},
+		}
+
+		for i, test := range []struct {
+			prefix     *string
+			delimiter  *string
+			versions   []types.ObjectVersion
+			delMarkers []types.DeleteMarkerEntry
+			prefixes   []string
+		}{
+			// "my-dir/" is rolled up into a common prefix
+			{
+				delimiter: getPtr("/"),
+				versions:  objVersions,
+				prefixes:  []string{dir},
+			},
+			// "my-dir/" itself and "my-dir/child" have no delimiter
+			// after the prefix
+			{
+				prefix:     &dir,
+				delimiter:  getPtr("/"),
+				versions:   append(dirVersions, childVersions...),
+				delMarkers: delMarkers,
+				prefixes:   []string{"my-dir/sub/"},
+			},
+			// "my-dir/sub" isn't a directory object
+			{
+				prefix:     &dir,
+				versions:   append(append(dirVersions, childVersions...), nestedVersions...),
+				delMarkers: delMarkers,
+			},
+		} {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket:    &bucket,
+				Prefix:    test.prefix,
+				Delimiter: test.delimiter,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("test case %d failed: %w", i, err)
+			}
+
+			if !compareVersions(test.versions, out.Versions) {
+				return fmt.Errorf("test case %d failed: expected the versions to be %v, instead got %v",
+					i, sprintVersions(test.versions), sprintVersions(out.Versions))
+			}
+			if !compareDelMarkers(test.delMarkers, out.DeleteMarkers) {
+				return fmt.Errorf("test case %d failed: expected the delete markers to be %v, instead got %v",
+					i, test.delMarkers, out.DeleteMarkers)
+			}
+			if !comparePrefixes(test.prefixes, out.CommonPrefixes) {
+				return fmt.Errorf("test case %d failed: expected the common prefixes to be %v, instead got %v",
+					i, test.prefixes, sprintPrefixes(out.CommonPrefixes))
+			}
 		}
 
 		return nil
@@ -162,7 +302,7 @@ func ListObjectVersions_list_multiple_object_versions(s *S3Conf) error {
 func ListObjectVersions_multiple_object_versions_truncated(s *S3Conf) error {
 	testName := "ListObjectVersions_multiple_object_versions_truncated"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj1, obj2, obj3 := "foo", "bar", "baz"
+		obj1, obj2, obj3, dir, child := "foo", "bar", "baz", "dir/", "dir/child"
 
 		obj1Versions, err := createObjVersions(s3client, bucket, obj1, 4)
 		if err != nil {
@@ -176,87 +316,85 @@ func ListObjectVersions_multiple_object_versions_truncated(s *S3Conf) error {
 		if err != nil {
 			return err
 		}
+		dirVersions, err := createObjVersions(s3client, bucket, dir, 4)
+		if err != nil {
+			return err
+		}
+		childVersions, err := createObjVersions(s3client, bucket, child, 3)
+		if err != nil {
+			return err
+		}
 
-		versions := append(append(obj2Versions, obj3Versions...), obj1Versions...)
+		versions := append(append(obj2Versions, obj3Versions...), dirVersions...)
+		versions = append(append(versions, childVersions...), obj1Versions...)
 		maxKeys := int32(5)
 
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
-			Bucket:  &bucket,
-			MaxKeys: &maxKeys,
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
+		// the pages end on noncurrent versions of "baz", "dir/" and "dir/child"
+		var keyMarker, versionIdMarker *string
+		for page := 0; page*int(maxKeys) < len(versions); page++ {
+			start := page * int(maxKeys)
+			end := min(start+int(maxKeys), len(versions))
+			truncated := end < len(versions)
 
-		if out.Name == nil {
-			return fmt.Errorf("expected the bucket name to be %v, instead got nil",
-				bucket)
-		}
-		if *out.Name != bucket {
-			return fmt.Errorf("expected the bucket name to be %v, instead got %v",
-				bucket, *out.Name)
-		}
-		if out.IsTruncated == nil || !*out.IsTruncated {
-			return fmt.Errorf("expected the output to be truncated")
-		}
-		if out.MaxKeys == nil {
-			return fmt.Errorf("expected the max-keys to be %v, instead got nil",
-				maxKeys)
-		}
-		if *out.MaxKeys != maxKeys {
-			return fmt.Errorf("expected the max-keys to be %v, instead got %v",
-				maxKeys, *out.MaxKeys)
-		}
-		if getString(out.NextKeyMarker) != getString(versions[maxKeys-1].Key) {
-			return fmt.Errorf("expected the NextKeyMarker to be %v, instead got %v",
-				getString(versions[maxKeys-1].Key), getString(out.NextKeyMarker))
-		}
-		if getString(out.NextVersionIdMarker) != getString(versions[maxKeys-1].VersionId) {
-			return fmt.Errorf("expected the NextVersionIdMarker to be %v, instead got %v",
-				getString(versions[maxKeys-1].VersionId), getString(out.NextVersionIdMarker))
-		}
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket:          &bucket,
+				MaxKeys:         &maxKeys,
+				KeyMarker:       keyMarker,
+				VersionIdMarker: versionIdMarker,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("page %v: %w", page, err)
+			}
 
-		if !compareVersions(versions[:maxKeys], out.Versions) {
-			return fmt.Errorf("expected the resulting object versions to be %v, instead got %v",
-				sprintVersions(versions[:maxKeys]), sprintVersions(out.Versions))
-		}
+			if out.Name == nil {
+				return fmt.Errorf("page %v: expected the bucket name to be %v, instead got nil",
+					page, bucket)
+			}
+			if *out.Name != bucket {
+				return fmt.Errorf("page %v: expected the bucket name to be %v, instead got %v",
+					page, bucket, *out.Name)
+			}
+			isTruncated := out.IsTruncated != nil && *out.IsTruncated
+			if isTruncated != truncated {
+				return fmt.Errorf("page %v: expected the output truncation to be %v, instead got %v",
+					page, truncated, isTruncated)
+			}
+			if out.MaxKeys == nil {
+				return fmt.Errorf("page %v: expected the max-keys to be %v, instead got nil",
+					page, maxKeys)
+			}
+			if *out.MaxKeys != maxKeys {
+				return fmt.Errorf("page %v: expected the max-keys to be %v, instead got %v",
+					page, maxKeys, *out.MaxKeys)
+			}
+			if getString(out.KeyMarker) != getString(keyMarker) {
+				return fmt.Errorf("page %v: expected the KeyMarker to be %v, instead got %v",
+					page, getString(keyMarker), getString(out.KeyMarker))
+			}
+			if getString(out.VersionIdMarker) != getString(versionIdMarker) {
+				return fmt.Errorf("page %v: expected the VersionIdMarker to be %v, instead got %v",
+					page, getString(versionIdMarker), getString(out.VersionIdMarker))
+			}
+			if truncated {
+				last := versions[end-1]
+				if getString(out.NextKeyMarker) != getString(last.Key) {
+					return fmt.Errorf("page %v: expected the NextKeyMarker to be %v, instead got %v",
+						page, getString(last.Key), getString(out.NextKeyMarker))
+				}
+				if getString(out.NextVersionIdMarker) != getString(last.VersionId) {
+					return fmt.Errorf("page %v: expected the NextVersionIdMarker to be %v, instead got %v",
+						page, getString(last.VersionId), getString(out.NextVersionIdMarker))
+				}
+			}
 
-		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
-		out, err = s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
-			Bucket:          &bucket,
-			KeyMarker:       out.NextKeyMarker,
-			VersionIdMarker: out.NextVersionIdMarker,
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
+			if !compareVersions(versions[start:end], out.Versions) {
+				return fmt.Errorf("page %v: expected the resulting object versions to be %v, instead got %v",
+					page, sprintVersions(versions[start:end]), sprintVersions(out.Versions))
+			}
 
-		if out.Name == nil {
-			return fmt.Errorf("expected the bucket name to be %v, instead got nil",
-				bucket)
-		}
-		if *out.Name != bucket {
-			return fmt.Errorf("expected the bucket name to be %v, instead got %v",
-				bucket, *out.Name)
-		}
-		if out.IsTruncated != nil && *out.IsTruncated {
-			return fmt.Errorf("expected the output not to be truncated")
-		}
-		if getString(out.KeyMarker) != getString(versions[maxKeys-1].Key) {
-			return fmt.Errorf("expected the KeyMarker to be %v, instead got %v",
-				getString(versions[maxKeys-1].Key), getString(out.KeyMarker))
-		}
-		if getString(out.VersionIdMarker) != getString(versions[maxKeys-1].VersionId) {
-			return fmt.Errorf("expected the VersionIdMarker to be %v, instead got %v",
-				getString(versions[maxKeys-1].VersionId), getString(out.VersionIdMarker))
-		}
-
-		if !compareVersions(versions[maxKeys:], out.Versions) {
-			return fmt.Errorf("expected the resulting object versions to be %v, instead got %v",
-				sprintVersions(versions[:maxKeys]), sprintVersions(out.Versions))
+			keyMarker, versionIdMarker = out.NextKeyMarker, out.NextVersionIdMarker
 		}
 
 		return nil
@@ -266,170 +404,180 @@ func ListObjectVersions_multiple_object_versions_truncated(s *S3Conf) error {
 func ListObjectVersions_with_delete_markers(s *S3Conf) error {
 	testName := "ListObjectVersions_with_delete_markers"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj := "my-obj"
-		versions, err := createObjVersions(s3client, bucket, obj, 1)
-		if err != nil {
-			return err
-		}
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			versions, err := createObjVersions(s3client, bucket, obj, 1)
+			if err != nil {
+				return err
+			}
 
-		versions[0].IsLatest = getBoolPtr(false)
+			versions[0].IsLatest = getBoolPtr(false)
 
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
-			Bucket: &bucket,
-			Key:    &obj,
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			delMarkers := []types.DeleteMarkerEntry{}
+			delMarkers = append(delMarkers, types.DeleteMarkerEntry{
+				Key:       &obj,
+				VersionId: out.VersionId,
+				IsLatest:  getBoolPtr(true),
+			})
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket: &bucket,
+				Prefix: &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if !compareVersions(versions, res.Versions) {
+				return fmt.Errorf("expected the resulting versions to be %v, instead got %v",
+					versions, res.Versions)
+			}
+			if !compareDelMarkers(res.DeleteMarkers, delMarkers) {
+				return fmt.Errorf("expected the resulting delete markers to be %v, instead got %v",
+					delMarkers, res.DeleteMarkers)
+			}
+
+			return nil
 		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		delMarkers := []types.DeleteMarkerEntry{}
-		delMarkers = append(delMarkers, types.DeleteMarkerEntry{
-			Key:       &obj,
-			VersionId: out.VersionId,
-			IsLatest:  getBoolPtr(true),
-		})
-
-		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
-		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
-			Bucket: &bucket,
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		if !compareVersions(versions, res.Versions) {
-			return fmt.Errorf("expected the resulting versions to be %v, instead got %v",
-				versions, res.Versions)
-		}
-		if !compareDelMarkers(res.DeleteMarkers, delMarkers) {
-			return fmt.Errorf("expected the resulting delete markers to be %v, instead got %v",
-				delMarkers, res.DeleteMarkers)
-		}
-
-		return nil
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
 func ListObjectVersions_containing_null_versionId_obj(s *S3Conf) error {
 	testName := "ListObjectVersions_containing_null_versionId_obj"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj := "my-obj"
-		versions, err := createObjVersions(s3client, bucket, obj, 3)
-		if err != nil {
-			return err
-		}
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			versions, err := createObjVersions(s3client, bucket, obj, 3)
+			if err != nil {
+				return err
+			}
 
-		err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusSuspended)
-		if err != nil {
-			return err
-		}
+			err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusSuspended)
+			if err != nil {
+				return err
+			}
 
-		objLgth := int64(543)
-		out, err := putObjectWithData(objLgth, &s3.PutObjectInput{
-			Bucket: &bucket,
-			Key:    &obj,
-		}, s3client)
-		if err != nil {
-			return err
-		}
+			objLgth := objDataLen(obj, 543)
+			out, err := putObjectWithData(objLgth, &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			}, s3client)
+			if err != nil {
+				return err
+			}
 
-		if out.res.VersionId != nil {
-			return fmt.Errorf("expected PutObject response to omit versionId, instead got %v",
-				getString(out.res.VersionId))
-		}
+			if out.res.VersionId != nil {
+				return fmt.Errorf("expected PutObject response to omit versionId, instead got %v",
+					getString(out.res.VersionId))
+			}
 
-		versions[0].IsLatest = getBoolPtr(false)
+			versions[0].IsLatest = getBoolPtr(false)
 
-		versions = append([]types.ObjectVersion{
-			{
-				ETag:         out.res.ETag,
-				IsLatest:     getBoolPtr(false),
-				Key:          &obj,
-				Size:         &objLgth,
-				VersionId:    &nullVersionId,
-				StorageClass: types.ObjectVersionStorageClassStandard,
-			},
-		}, versions...)
+			versions = append([]types.ObjectVersion{
+				{
+					ETag:         out.res.ETag,
+					IsLatest:     getBoolPtr(false),
+					Key:          &obj,
+					Size:         &objLgth,
+					VersionId:    &nullVersionId,
+					StorageClass: types.ObjectVersionStorageClassStandard,
+				},
+			}, versions...)
 
-		err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
-		if err != nil {
-			return err
-		}
+			err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
+			if err != nil {
+				return err
+			}
 
-		newVersions, err := createObjVersions(s3client, bucket, obj, 4)
-		if err != nil {
-			return err
-		}
+			newVersions, err := createObjVersions(s3client, bucket, obj, 4)
+			if err != nil {
+				return err
+			}
 
-		versions = append(newVersions, versions...)
+			versions = append(newVersions, versions...)
 
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
-			Bucket: &bucket,
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+				Bucket: &bucket,
+				Prefix: &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			if !compareVersions(versions, res.Versions) {
+				return fmt.Errorf("expected the listed object versions to be %v, instead got %v",
+					versions, res.Versions)
+			}
+
+			return nil
 		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		if !compareVersions(versions, res.Versions) {
-			return fmt.Errorf("expected the listed object versions to be %v, instead got %v",
-				versions, res.Versions)
-		}
-
-		return nil
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
 func ListObjectVersions_single_null_versionId_object(s *S3Conf) error {
 	testName := "ListObjectVersions_single_null_versionId_object"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		obj, objLgth := "my-obj", int64(890)
-		out, err := putObjectWithData(objLgth, &s3.PutObjectInput{
-			Bucket: &bucket,
-			Key:    &obj,
-		}, s3client)
-		if err != nil {
-			return err
-		}
+		// the objects are put before versioning is enabled and are
+		// listed in this order
+		objs, objLgth := []string{"my-dir/", "my-obj"}, int64(890)
+		versions := []types.ObjectVersion{}
+		for _, obj := range objs {
+			size := objDataLen(obj, objLgth)
+			out, err := putObjectWithData(size, &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			}, s3client)
+			if err != nil {
+				return err
+			}
 
-		err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
-		if err != nil {
-			return err
-		}
-
-		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-		res, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
-			Bucket: &bucket,
-			Key:    &obj,
-		})
-		cancel()
-		if err != nil {
-			return err
-		}
-
-		versions := []types.ObjectVersion{
-			{
+			versions = append(versions, types.ObjectVersion{
 				ETag:         out.res.ETag,
 				Key:          &obj,
 				StorageClass: types.ObjectVersionStorageClassStandard,
 				IsLatest:     getBoolPtr(false),
-				Size:         &objLgth,
+				Size:         &size,
 				VersionId:    &nullVersionId,
-			},
+			})
 		}
-		delMarkers := []types.DeleteMarkerEntry{
-			{
+
+		err := putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
+		if err != nil {
+			return err
+		}
+
+		delMarkers := []types.DeleteMarkerEntry{}
+		for _, obj := range objs {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			delMarkers = append(delMarkers, types.DeleteMarkerEntry{
 				IsLatest:  getBoolPtr(true),
 				Key:       &obj,
 				VersionId: res.VersionId,
-			},
+			})
 		}
 
-		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		resp, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
 			Bucket: &bucket,
 		})
@@ -454,7 +602,7 @@ func ListObjectVersions_single_null_versionId_object(s *S3Conf) error {
 func ListObjectVersions_checksum(s *S3Conf) error {
 	testName := "ListObjectVersions_checksum"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		versions := []types.ObjectVersion{}
+		versions, dirVersions := []types.ObjectVersion{}, []types.ObjectVersion{}
 		for i, algo := range types.ChecksumAlgorithmCrc32.Values() {
 			vers, err := createObjVersions(s3client, bucket, fmt.Sprintf("obj-%v", i), 1, withChecksumAlgo(algo))
 			if err != nil {
@@ -462,8 +610,18 @@ func ListObjectVersions_checksum(s *S3Conf) error {
 			}
 
 			versions = append(versions, vers...)
+
+			vers, err = createObjVersions(s3client, bucket, fmt.Sprintf("dir-%v/", i), 1, withChecksumAlgo(algo))
+			if err != nil {
+				return err
+			}
+
+			dirVersions = append(dirVersions, vers...)
 		}
 
+		// the directory objects are listed before the regular objects
+		versions = append(dirVersions, versions...)
+
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
 			Bucket: &bucket,

tests/integration/group-tests.go

@@ -994,6 +994,12 @@ func TestPosix(ts *TestState) {
 	ts.Run(DeleteObject_name_too_long)
 	ts.Run(CopyObject_overwrite_same_dir_object)
 	ts.Run(CopyObject_overwrite_same_file_object)
+	ts.Run(CompleteMultipartUpload_overwrite_dir_obj)
+	if ts.conf.versioningEnabled {
+		ts.Run(CompleteMultipartUpload_overwrite_dir_obj_delete_marker)
+	}
+	ts.Run(ObjectTagging_trailing_slash_counterpart)
+	ts.Run(ObjectLock_trailing_slash_counterpart)
 	ts.Run(DeleteObject_directory_not_empty)
 	if !ts.conf.windowsTests {
 		ts.Run(PutObject_race_with_delete)
@@ -1098,6 +1104,9 @@ func TestScoutfs(ts *TestState) {
 	ts.Run(DeleteObject_name_too_long)
 	ts.Run(CopyObject_overwrite_same_dir_object)
 	ts.Run(CopyObject_overwrite_same_file_object)
+	ts.Run(CompleteMultipartUpload_overwrite_dir_obj)
+	ts.Run(ObjectTagging_trailing_slash_counterpart)
+	ts.Run(ObjectLock_trailing_slash_counterpart)
 	ts.Run(DeleteObject_directory_not_empty)
 }
 
@@ -1945,12 +1954,16 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_PutObject_null_versionId_obj)
 	ts.Run(Versioning_PutObject_overwrite_null_versionId_obj)
 	ts.Run(Versioning_PutObject_success)
+	ts.Run(Versioning_PutObject_dir_object_new_version_resets_attributes)
 	// CopyObject action
 	ts.Run(Versioning_CopyObject_invalid_versionId)
 	ts.Run(Versioning_CopyObject_encoded_versionid_separator_invalid_versionId)
 	ts.Run(Versioning_CopyObject_success)
 	ts.Run(Versioning_CopyObject_non_existing_version_id)
 	ts.Run(Versioning_CopyObject_from_an_object_version)
+	ts.Run(Versioning_CopyObject_from_a_delete_marker)
+	ts.Run(Versioning_CopyObject_to_itself)
+	ts.Run(Versioning_CopyObject_to_itself_from_the_current_version)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_CopyObject_special_chars)
 	}
@@ -1961,7 +1974,10 @@ func TestVersioning(ts *TestState) {
 	}
 	ts.Run(Versioning_HeadObject_invalid_parent)
 	ts.Run(Versioning_HeadObject_success)
+	ts.Run(Versioning_HeadObject_dir_object_versions)
 	ts.Run(Versioning_HeadObject_without_versionId)
+	ts.Run(Versioning_HeadObject_null_version_without_versionId)
+	ts.Run(Versioning_HeadObject_null_versionId_obj)
 	ts.Run(Versioning_HeadObject_delete_marker)
 	// GetObject action
 	ts.Run(Versioning_GetObject_invalid_versionId)
@@ -1970,6 +1986,8 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_GetObject_delete_marker_without_versionId)
 	ts.Run(Versioning_GetObject_delete_marker)
 	ts.Run(Versioning_GetObject_null_versionId_obj)
+	ts.Run(Versioning_GetObject_null_version_without_versionId)
+	ts.Run(Versioning_unversioned_bucket_omits_versionId)
 	// object tagging actions
 	ts.Run(Versioning_PutObjectTagging_invalid_versionId)
 	ts.Run(Versioning_PutObjectTagging_non_existing_object_version)
@@ -1979,6 +1997,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_DeleteObjectTagging_invalid_versionId)
 	ts.Run(Versioning_DeleteObjectTagging_non_existing_object_version)
 	ts.Run(Versioning_PutGetDeleteObjectTagging_success)
+	ts.Run(Versioning_ObjectTagging_trailing_slash_counterpart)
 	// GetObjectAttributes action
 	ts.Run(Versioning_GetObjectAttributes_invalid_versionId)
 	ts.Run(Versioning_GetObjectAttributes_object_version)
@@ -1986,9 +2005,14 @@ func TestVersioning(ts *TestState) {
 	// DeleteObject actions
 	ts.Run(Versioning_DeleteObject_invalid_versionId)
 	ts.Run(Versioning_DeleteObject_delete_object_version)
+	ts.Run(Versioning_DeleteObject_dir_object_latest_version)
+	ts.Run(Versioning_DeleteObject_latest_version_with_null_version)
 	ts.Run(Versioning_DeleteObject_non_existing_object)
+	ts.Run(Versioning_DeleteObject_implicit_dir)
+	ts.Run(Versioning_DeleteObject_trailing_slash_counterpart)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_DeleteObject_delete_a_delete_marker)
+		ts.Run(Versioning_DeleteObject_dir_object_with_children)
 	}
 	ts.Run(Versioning_Delete_null_versionId_object)
 	ts.Run(Versioning_DeleteObject_nested_dir_object)
@@ -2001,6 +2025,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(ListObjectVersions_negative_max_keys)
 	ts.Run(ListObjectVersions_list_single_object_versions)
 	ts.Run(ListObjectVersions_list_multiple_object_versions)
+	ts.Run(ListObjectVersions_dir_object_with_children)
 	ts.Run(ListObjectVersions_multiple_object_versions_truncated)
 	ts.Run(ListObjectVersions_with_delete_markers)
 	ts.Run(ListObjectVersions_containing_null_versionId_obj)
@@ -2013,6 +2038,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId)
 	ts.Run(Versioning_UploadPartCopy_non_existing_versionId)
 	ts.Run(Versioning_UploadPartCopy_from_an_object_version)
+	ts.Run(Versioning_UploadPartCopy_from_a_delete_marker)
 	// Object lock configuration
 	ts.Run(Versioning_object_lock_not_enabled_on_bucket_creation)
 	ts.Run(Versioning_Enable_object_lock)
@@ -2033,6 +2059,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_Put_GetObjectLegalHold_success)
 	// WORM protection
 	ts.Run(Versioning_WORM_obj_version_locked_with_legal_hold)
+	ts.Run(Versioning_WORM_dir_object_lock_headers)
 	ts.Run(Versioning_WORM_obj_version_locked_with_governance_retention)
 	ts.Run(Versioning_WORM_obj_version_locked_with_compliance_retention)
 	ts.Run(Versioning_WORM_delete_marker_locked_object_legal_hold)
@@ -2040,10 +2067,13 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_WORM_delete_marker_locked_object_compliance_retention)
 	ts.Run(Versioning_WORM_PutObject_overwrite_locked_object)
 	ts.Run(Versioning_WORM_CopyObject_overwrite_locked_object)
+	ts.Run(Versioning_WORM_CopyObject_to_itself_locked_object)
 	ts.Run(Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_WORM_remove_delete_marker_under_bucket_default_retention)
 	}
+	ts.Run(Versioning_WORM_trailing_slash_counterpart)
+	ts.Run(Versioning_WORM_null_version_locked_with_legal_hold)
 	// Concurrent requests
 	// Versioninig_concurrent_upload_object
 	ts.Run(Versioning_AccessControl_GetObjectVersion)
@@ -2990,6 +3020,10 @@ func GetIntTests() IntTests {
 		"DeleteObject_name_too_long":                                                       DeleteObject_name_too_long,
 		"CopyObject_overwrite_same_dir_object":                                             CopyObject_overwrite_same_dir_object,
 		"CopyObject_overwrite_same_file_object":                                            CopyObject_overwrite_same_file_object,
+		"CompleteMultipartUpload_overwrite_dir_obj":                                        CompleteMultipartUpload_overwrite_dir_obj,
+		"CompleteMultipartUpload_overwrite_dir_obj_delete_marker":                          CompleteMultipartUpload_overwrite_dir_obj_delete_marker,
+		"ObjectTagging_trailing_slash_counterpart":                                         ObjectTagging_trailing_slash_counterpart,
+		"ObjectLock_trailing_slash_counterpart":                                            ObjectLock_trailing_slash_counterpart,
 		"DeleteObject_non_existing_dir_object":                                             DeleteObject_non_existing_dir_object,
 		"DeleteObject_directory_object":                                                    DeleteObject_directory_object,
 		"DeleteObject_success":                                                             DeleteObject_success,
@@ -3473,23 +3507,32 @@ func GetIntTests() IntTests {
 		"Versioning_PutObject_null_versionId_obj":                                          Versioning_PutObject_null_versionId_obj,
 		"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_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,
 		"Versioning_CopyObject_non_existing_version_id":                                    Versioning_CopyObject_non_existing_version_id,
 		"Versioning_CopyObject_from_an_object_version":                                     Versioning_CopyObject_from_an_object_version,
+		"Versioning_CopyObject_from_a_delete_marker":                                       Versioning_CopyObject_from_a_delete_marker,
+		"Versioning_CopyObject_to_itself":                                                  Versioning_CopyObject_to_itself,
+		"Versioning_CopyObject_to_itself_from_the_current_version":                         Versioning_CopyObject_to_itself_from_the_current_version,
 		"Versioning_CopyObject_special_chars":                                              Versioning_CopyObject_special_chars,
 		"Versioning_HeadObject_invalid_versionId":                                          Versioning_HeadObject_invalid_versionId,
 		"Versioning_HeadObject_non_existing_object_version":                                Versioning_HeadObject_non_existing_object_version,
 		"Versioning_HeadObject_invalid_parent":                                             Versioning_HeadObject_invalid_parent,
 		"Versioning_HeadObject_success":                                                    Versioning_HeadObject_success,
+		"Versioning_HeadObject_dir_object_versions":                                        Versioning_HeadObject_dir_object_versions,
 		"Versioning_HeadObject_without_versionId":                                          Versioning_HeadObject_without_versionId,
+		"Versioning_HeadObject_null_versionId_obj":                                         Versioning_HeadObject_null_versionId_obj,
+		"Versioning_HeadObject_null_version_without_versionId":                             Versioning_HeadObject_null_version_without_versionId,
 		"Versioning_HeadObject_delete_marker":                                              Versioning_HeadObject_delete_marker,
 		"Versioning_GetObject_invalid_versionId":                                           Versioning_GetObject_invalid_versionId,
 		"Versioning_GetObject_non_existing_object_version":                                 Versioning_GetObject_non_existing_object_version,
 		"Versioning_GetObject_success":                                                     Versioning_GetObject_success,
 		"Versioning_GetObject_delete_marker_without_versionId":                             Versioning_GetObject_delete_marker_without_versionId,
 		"Versioning_GetObject_delete_marker":                                               Versioning_GetObject_delete_marker,
+		"Versioning_unversioned_bucket_omits_versionId":                                    Versioning_unversioned_bucket_omits_versionId,
+		"Versioning_GetObject_null_version_without_versionId":                              Versioning_GetObject_null_version_without_versionId,
 		"Versioning_GetObject_null_versionId_obj":                                          Versioning_GetObject_null_versionId_obj,
 		"Versioning_PutObjectTagging_invalid_versionId":                                    Versioning_PutObjectTagging_invalid_versionId,
 		"Versioning_PutObjectTagging_non_existing_object_version":                          Versioning_PutObjectTagging_non_existing_object_version,
@@ -3499,13 +3542,19 @@ func GetIntTests() IntTests {
 		"Versioning_DeleteObjectTagging_invalid_versionId":                                 Versioning_DeleteObjectTagging_invalid_versionId,
 		"Versioning_DeleteObjectTagging_non_existing_object_version":                       Versioning_DeleteObjectTagging_non_existing_object_version,
 		"Versioning_PutGetDeleteObjectTagging_success":                                     Versioning_PutGetDeleteObjectTagging_success,
+		"Versioning_ObjectTagging_trailing_slash_counterpart":                              Versioning_ObjectTagging_trailing_slash_counterpart,
 		"Versioning_GetObjectAttributes_invalid_versionId":                                 Versioning_GetObjectAttributes_invalid_versionId,
 		"Versioning_GetObjectAttributes_object_version":                                    Versioning_GetObjectAttributes_object_version,
 		"Versioning_GetObjectAttributes_delete_marker":                                     Versioning_GetObjectAttributes_delete_marker,
 		"Versioning_DeleteObject_invalid_versionId":                                        Versioning_DeleteObject_invalid_versionId,
 		"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_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,
+		"Versioning_DeleteObject_trailing_slash_counterpart":                               Versioning_DeleteObject_trailing_slash_counterpart,
 		"Versioning_DeleteObject_delete_a_delete_marker":                                   Versioning_DeleteObject_delete_a_delete_marker,
+		"Versioning_DeleteObject_dir_object_with_children":                                 Versioning_DeleteObject_dir_object_with_children,
 		"Versioning_Delete_null_versionId_object":                                          Versioning_Delete_null_versionId_object,
 		"Versioning_DeleteObject_nested_dir_object":                                        Versioning_DeleteObject_nested_dir_object,
 		"Versioning_DeleteObject_non_existing_objects":                                     Versioning_DeleteObject_non_existing_objects,
@@ -3516,6 +3565,7 @@ func GetIntTests() IntTests {
 		"ListObjectVersions_negative_max_keys":                                             ListObjectVersions_negative_max_keys,
 		"ListObjectVersions_list_single_object_versions":                                   ListObjectVersions_list_single_object_versions,
 		"ListObjectVersions_list_multiple_object_versions":                                 ListObjectVersions_list_multiple_object_versions,
+		"ListObjectVersions_dir_object_with_children":                                      ListObjectVersions_dir_object_with_children,
 		"ListObjectVersions_multiple_object_versions_truncated":                            ListObjectVersions_multiple_object_versions_truncated,
 		"ListObjectVersions_with_delete_markers":                                           ListObjectVersions_with_delete_markers,
 		"ListObjectVersions_containing_null_versionId_obj":                                 ListObjectVersions_containing_null_versionId_obj,
@@ -3527,6 +3577,7 @@ func GetIntTests() IntTests {
 		"Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId":          Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId,
 		"Versioning_UploadPartCopy_non_existing_versionId":                                 Versioning_UploadPartCopy_non_existing_versionId,
 		"Versioning_UploadPartCopy_from_an_object_version":                                 Versioning_UploadPartCopy_from_an_object_version,
+		"Versioning_UploadPartCopy_from_a_delete_marker":                                   Versioning_UploadPartCopy_from_a_delete_marker,
 		"Versioning_object_lock_not_enabled_on_bucket_creation":                            Versioning_object_lock_not_enabled_on_bucket_creation,
 		"Versioning_Enable_object_lock":                                                    Versioning_Enable_object_lock,
 		"Versioning_status_switch_to_suspended_with_object_lock":                           Versioning_status_switch_to_suspended_with_object_lock,
@@ -3543,6 +3594,7 @@ func GetIntTests() IntTests {
 		"Versioning_PutGetObjectLegalHold_delete_marker":                                   Versioning_PutGetObjectLegalHold_delete_marker,
 		"Versioning_Put_GetObjectLegalHold_success":                                        Versioning_Put_GetObjectLegalHold_success,
 		"Versioning_WORM_obj_version_locked_with_legal_hold":                               Versioning_WORM_obj_version_locked_with_legal_hold,
+		"Versioning_WORM_dir_object_lock_headers":                                          Versioning_WORM_dir_object_lock_headers,
 		"Versioning_WORM_obj_version_locked_with_governance_retention":                     Versioning_WORM_obj_version_locked_with_governance_retention,
 		"Versioning_WORM_obj_version_locked_with_compliance_retention":                     Versioning_WORM_obj_version_locked_with_compliance_retention,
 		"Versioning_WORM_delete_marker_locked_object_legal_hold":                           Versioning_WORM_delete_marker_locked_object_legal_hold,
@@ -3550,8 +3602,11 @@ func GetIntTests() IntTests {
 		"Versioning_WORM_delete_marker_locked_object_compliance_retention":                 Versioning_WORM_delete_marker_locked_object_compliance_retention,
 		"Versioning_WORM_PutObject_overwrite_locked_object":                                Versioning_WORM_PutObject_overwrite_locked_object,
 		"Versioning_WORM_CopyObject_overwrite_locked_object":                               Versioning_WORM_CopyObject_overwrite_locked_object,
+		"Versioning_WORM_CopyObject_to_itself_locked_object":                               Versioning_WORM_CopyObject_to_itself_locked_object,
 		"Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object":                  Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object,
 		"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_AccessControl_GetObjectVersion":                                        Versioning_AccessControl_GetObjectVersion,
 		"Versioning_AccessControl_HeadObjectVersion":                                       Versioning_AccessControl_HeadObjectVersion,
 		"Versioning_AccessControl_object_tagging_policy":                                   Versioning_AccessControl_object_tagging_policy,

tests/integration/posix.go

@@ -17,6 +17,7 @@ package integration
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -222,6 +223,255 @@ func CopyObject_overwrite_same_file_object(s *S3Conf) error {
 	})
 }
 
+func CompleteMultipartUpload_overwrite_dir_obj(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_overwrite_dir_obj"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dir, obj := "foo/", "foo"
+		_, err := putObjects(s3client, []string{dir}, bucket)
+		if err != nil {
+			return err
+		}
+
+		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 := checkApiErr(err, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)); err != nil {
+			return err
+		}
+
+		// the directory object isn't taken for the object of a completed upload
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: getPtr("non-existing-upload-id"),
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{ETag: parts[0].ETag, PartNumber: parts[0].PartNumber},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the failed upload can still be completed or aborted
+		return checkAndAbortUpload(s3client, bucket, obj, *mp.UploadId)
+	})
+}
+
+func CompleteMultipartUpload_overwrite_dir_obj_delete_marker(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_overwrite_dir_obj_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dir, obj := "foo/", "foo"
+		versions, err := createObjVersions(s3client, bucket, dir, 1)
+		if err != nil {
+			return err
+		}
+		versions[0].IsLatest = getPtr(false)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+		delMarkers := []types.DeleteMarkerEntry{
+			{Key: &dir, VersionId: out.VersionId, IsLatest: getPtr(true)},
+		}
+
+		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 := checkApiErr(err, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)); err != nil {
+			return err
+		}
+
+		// the delete marker of the directory object is left in place
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(versions, res.Versions) {
+			return fmt.Errorf("expected the versions to be %v, instead got %v",
+				versions, res.Versions)
+		}
+		if !compareDelMarkers(delMarkers, res.DeleteMarkers) {
+			return fmt.Errorf("expected the delete markers to be %v, instead got %v",
+				delMarkers, res.DeleteMarkers)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err := checkSdkApiErr(err, "NotFound"); err != nil {
+			return err
+		}
+
+		return checkAndAbortUpload(s3client, bucket, obj, *mp.UploadId)
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+func ObjectTagging_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "ObjectTagging_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagSet := []types.Tag{{Key: getPtr("key"), Value: getPtr("value")}}
+
+		// a key and the same key with a trailing slash have one path: the
+		// object of one of them isn't an object of the other
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			_, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:  &bucket,
+				Key:     &obj,
+				Tagging: getPtr("key=value"),
+			}, s3client)
+			if err != nil {
+				return err
+			}
+
+			err = checkObjectTaggingErr(s3client, bucket, other, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+			if !areTagsSame(res.TagSet, tagSet) {
+				return fmt.Errorf("%v: expected the tag set to be %v, instead got %v",
+					obj, tagSet, res.TagSet)
+			}
+		}
+
+		// neither key names the parent directory of an object
+		_, err := putObjects(s3client, []string{"my-parent/obj"}, bucket)
+		if err != nil {
+			return err
+		}
+		for _, key := range []string{"my-parent/", "my-parent"} {
+			err := checkObjectTaggingErr(s3client, bucket, key, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", key, err)
+			}
+		}
+
+		return nil
+	})
+}
+
+func ObjectLock_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "ObjectLock_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		rDate := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
+		lockedObjs := []objToDelete{}
+
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			_, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:                    &bucket,
+				Key:                       &obj,
+				ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+				ObjectLockMode:            types.ObjectLockModeGovernance,
+				ObjectLockRetainUntilDate: &rDate,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			lockedObjs = append(lockedObjs, objToDelete{key: obj, removeOnlyLeglHold: true})
+
+			err = checkObjectLockErr(s3client, bucket, other, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			// the lock of the object doesn't protect the other key
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &other,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			err = checkObjectLock(s3client, bucket, obj, "", rDate)
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+		}
+
+		return cleanupLockedObjects(s3client, bucket, lockedObjs)
+	}, withLock())
+}
+
 // PutObject_race_with_delete tests the race between PutObject and DeleteObject
 // in the same subdirectory.
 // One goroutine sequentially puts "race-dir/0.txt" … "race-dir/N-1.txt".

tests/integration/utils.go

@@ -1234,22 +1234,24 @@ func putObjectWithData(lgth int64, input *s3.PutObjectInput, client *s3.Client,
 
 	var csum [32]byte
 	var data []byte
-	if input.Body == nil && lgth != 0 {
-		data = make([]byte, lgth)
-		rand.Read(data)
+	if input.Body == nil {
+		if lgth != 0 {
+			data = make([]byte, lgth)
+			rand.Read(data)
 
-		csum = sha256.Sum256(data)
-		if cfg.checksumAlgorithm != "" {
-			hasher, err := NewHasher(cfg.checksumAlgorithm)
-			if err != nil {
-				return nil, err
+			if cfg.checksumAlgorithm != "" {
+				hasher, err := NewHasher(cfg.checksumAlgorithm)
+				if err != nil {
+					return nil, err
+				}
+
+				hasher.Write(data)
+				sum := base64.StdEncoding.EncodeToString(hasher.Sum(nil))
+				setPutObjectChecksum(input, cfg.checksumAlgorithm, &sum)
 			}
-
-			hasher.Write(data)
-			sum := base64.StdEncoding.EncodeToString(hasher.Sum(nil))
-			setPutObjectChecksum(input, cfg.checksumAlgorithm, &sum)
+			input.Body = bytes.NewReader(data)
 		}
-		input.Body = bytes.NewReader(data)
+		csum = sha256.Sum256(data)
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), longTimeout)
@@ -2298,10 +2300,10 @@ func createObjVersions(client *s3.Client, bucket, object string, count int, opts
 	versions := []types.ObjectVersion{}
 	for i := range count {
 		rNumber, err := rand.Int(rand.Reader, big.NewInt(100000))
-		dataLength := rNumber.Int64()
 		if err != nil {
 			return nil, err
 		}
+		dataLength := objDataLen(object, rNumber.Int64())
 
 		r, err := putObjectWithData(dataLength, &s3.PutObjectInput{
 			Bucket: &bucket,
@@ -2373,6 +2375,49 @@ func createObjVersions(client *s3.Client, bucket, object string, count int, opts
 	return versions, nil
 }
 
+// createDeleteMarker deletes object without a version id, making the
+// resulting delete marker the current version, and returns its version id.
+func createDeleteMarker(client *s3.Client, bucket, object string) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	out, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
+		Bucket: &bucket,
+		Key:    &object,
+	})
+	cancel()
+	if err != nil {
+		return "", err
+	}
+	if out.DeleteMarker == nil || !*out.DeleteMarker {
+		return "", fmt.Errorf("expected a delete marker to be created for %v", object)
+	}
+	if getString(out.VersionId) == "" {
+		return "", fmt.Errorf("expected non empty delete marker versionId for %v", object)
+	}
+
+	return *out.VersionId, nil
+}
+
+// objDataLen returns the data length to upload for key: a directory
+// object can't hold data
+func objDataLen(key string, lgth int64) int64 {
+	if strings.HasSuffix(key, "/") {
+		return 0
+	}
+	return lgth
+}
+
+// forEachKey runs fn for each of the keys and prefixes a returned error
+// with the key it failed for. Tests use it to run the same scenario for
+// a regular object and a directory object.
+func forEachKey(keys []string, fn func(key string) error) error {
+	for _, key := range keys {
+		if err := fn(key); err != nil {
+			return fmt.Errorf("%v: %w", key, err)
+		}
+	}
+	return nil
+}
+
 // ReverseSlice reverses a slice of any type
 func reverseSlice[T any](s []T) []T {
 	for i, j := 0, len(s)-1; i < j; i, j = i+1, j-1 {
@@ -3966,3 +4011,170 @@ func checkDeleteObjectsErrsInOrder(got []types.Error, want []keyDenial) error {
 	}
 	return nil
 }
+
+// checkObjectTaggingErr checks that getting, putting and deleting the
+// tagging of the version versionId of key fail with expected. An empty
+// versionId selects the current version.
+func checkObjectTaggingErr(client *s3.Client, bucket, key, versionId string, expected s3err.S3Error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object tagging: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectTagging(ctx, &s3.PutObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{{Key: getPtr("other-key"), Value: getPtr("other-value")}},
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object tagging: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.DeleteObjectTagging(ctx, &s3.DeleteObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("delete object tagging: %w", err)
+	}
+
+	return nil
+}
+
+// checkObjectLockErr checks that getting and putting the legal hold and the
+// retention of the version versionId of key fail with expected. An empty
+// versionId selects the current version.
+func checkObjectLockErr(client *s3.Client, bucket, key, versionId string, expected s3err.S3Error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object legal hold: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectLegalHold(ctx, &s3.PutObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		LegalHold: &types.ObjectLockLegalHold{
+			Status: types.ObjectLockLegalHoldStatusOff,
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object legal hold: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.GetObjectRetention(ctx, &s3.GetObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object retention: %w", err)
+	}
+
+	rDate := time.Now().Add(time.Hour * 2)
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectRetention(ctx, &s3.PutObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		Retention: &types.ObjectLockRetention{
+			Mode:            types.ObjectLockRetentionModeGovernance,
+			RetainUntilDate: &rDate,
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object retention: %w", err)
+	}
+
+	return nil
+}
+
+// checkObjectLock checks that the version versionId of key is under legal
+// hold and has a governance retention until rDate. An empty versionId
+// selects the current version.
+func checkObjectLock(client *s3.Client, bucket, key, versionId string, rDate time.Time) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	lHold, err := client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if lHold.LegalHold == nil || lHold.LegalHold.Status != types.ObjectLockLegalHoldStatusOn {
+		return fmt.Errorf("expected the legal hold status to be %q, instead got %v",
+			types.ObjectLockLegalHoldStatusOn, lHold.LegalHold)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	ret, err := client.GetObjectRetention(ctx, &s3.GetObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if ret.Retention == nil || ret.Retention.Mode != types.ObjectLockRetentionModeGovernance ||
+		ret.Retention.RetainUntilDate == nil || ret.Retention.RetainUntilDate.Unix() != rDate.Unix() {
+		return fmt.Errorf("expected a %q retention until %v, instead got %+v",
+			types.ObjectLockRetentionModeGovernance, rDate.Format(time.RFC3339), ret.Retention)
+	}
+
+	return nil
+}
+
+// checkAndAbortUpload checks that the upload uploadId of key is the only
+// multipart upload in the bucket and aborts it
+func checkAndAbortUpload(client *s3.Client, bucket, key, uploadId string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	res, err := client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+		Bucket: &bucket,
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if len(res.Uploads) != 1 || getString(res.Uploads[0].Key) != key ||
+		getString(res.Uploads[0].UploadId) != uploadId {
+		return fmt.Errorf("expected the upload %v of %v to be listed, instead got %+v",
+			uploadId, key, res.Uploads)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
+		Bucket:   &bucket,
+		Key:      &key,
+		UploadId: &uploadId,
+	})
+	cancel()
+	return err
+}