feat: add versioning support for directory objects in posix

Directory objects (keys ending in `/`) are now versioned in versioning-enabled posix buckets like regular objects. The directory is kept in place across versions and carries the `version-id` and `delete-marker` attributes, while older versions are stored as empty files in the versioning directory. Only the attributes listed in `dirObjectAttrs` are copied into versions, replaced on a new `PutObject`, and restored when the latest version is deleted, so attributes the directory carries as a parent of other objects are left untouched. `PutObject` on a directory object now snapshots the current version, starts the new version with a clean attribute set and returns a `VersionId`. `DeleteObject` creates delete markers for directory objects and, when deleting the latest version, restores the previous one onto the directory, or removes the directory (or keeps it as a plain parent of its children) when no versions are left. `ListObjectVersions` lists directory object versions and delete markers and uses the key with the trailing slash for `NextKeyMarker` and marker comparisons, `ListObjects` hides directory delete markers and `GetObject` returns the directory object version id. `PutObject` on directory objects now applies the object lock legal hold and retention headers, and `CopyObject` and `UploadPartCopy` accept a directory object as a 0-byte copy source. The versioning integration tests are extended to run their scenarios for directory objects as well, with new tests for directory-object-specific behavior.
2026-09-25 17:34:29 +00:00 · 2026-09-19 02:10:34 +04:00
parent 1246a08cfa
commit 23317c766c
6 changed files with 4063 additions and 2538 deletions
@@ -1264,6 +1264,81 @@ func isRemovableAttr(attr string) bool {
 	}
 }

+// dirObjectAttrs are the attributes that make up the state of a directory
+// object. The directory is kept across versions and is also the parent of
+// other objects, so it may carry attributes that don't belong to the object:
+// only these are copied into, restored from and cleared between versions.
+// An attribute stored on objects must also be listed here.
+var dirObjectAttrs = []string{
+	etagkey,
+	checksumsKey,
+	contentTypeHdr,
+	contentEncHdr,
+	contentLangHdr,
+	contentDispHdr,
+	cacheCtrlHdr,
+	expiresHdr,
+	websiteRedirectHdr,
+	metadataHdr,
+	tagHdr,
+	objectLegalHoldKey,
+	objectRetentionKey,
+	versionIdKey,
+	deleteMarkerKey,
+}
+
+// isDirObject reports whether the directory at bucket/key is a directory
+// object: only directories created with a put carry an etag
+func (p *Posix) isDirObject(bucket, key string) (bool, error) {
+	_, err := p.meta.RetrieveAttribute(nil, bucket, key, etagkey)
+	if errors.Is(err, meta.ErrNoSuchKey) || errors.Is(err, fs.ErrNotExist) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("get dir etag: %w", err)
+	}
+	return true, nil
+}
+
+// isLiveDirObject reports whether fi, the entry at the path of the
+// directory object key, is a directory object
+func (p *Posix) isLiveDirObject(fi os.FileInfo, bucket, key string) (bool, error) {
+	if !fi.IsDir() {
+		return false, nil
+	}
+	return p.isDirObject(bucket, key)
+}
+
+// clearDirObjectAttrs removes the directory object attributes, including
+// legacy metadata attributes, from the directory at bucket/key. The etag is
+// kept: it marks the directory as an object, so a failure before the new
+// attributes are stored leaves the object and its versions visible.
+func (p *Posix) clearDirObjectAttrs(bucket, key string) error {
+	attrs, err := p.meta.ListAttributes(bucket, key)
+	if err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return fmt.Errorf("list object attributes: %w", err)
+	}
+	for _, attr := range attrs {
+		if isValidMeta(attr) {
+			err := p.meta.DeleteAttribute(bucket, key, attr)
+			if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
+				return fmt.Errorf("remove %v attribute: %w", attr, err)
+			}
+		}
+	}
+
+	for _, attr := range dirObjectAttrs {
+		if attr == etagkey {
+			continue
+		}
+		err := p.meta.DeleteAttribute(bucket, key, attr)
+		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) && !errors.Is(err, fs.ErrNotExist) {
+			return fmt.Errorf("remove %v attribute: %w", attr, err)
+		}
+	}
+	return nil
+}
+
 // Creates a new copy(version) of an object in the versioning directory
 func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Account, removeAttributes bool) (versionPath string, err error) {
 	sf, err := os.Open(p.ObjectPath(bucket, key))
@@ -1283,9 +1358,19 @@ func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Accoun
 		versionId = nullVersionId
 	}

-	attrs, err := p.meta.ListAttributes(bucket, key)
-	if err != nil {
-		return versionPath, fmt.Errorf("load object attributes: %w", err)
+	// a directory object version is an empty file carrying the
+	// directory object attributes
+	isDir := strings.HasSuffix(key, "/")
+	attrs := dirObjectAttrs
+	if isDir {
+		size = 0
+		// store legacy metadata attributes as the metadata attribute
+		p.loadObjectMetadata(sf, bucket, key)
+	} else {
+		attrs, err = p.meta.ListAttributes(bucket, key)
+		if err != nil {
+			return versionPath, fmt.Errorf("load object attributes: %w", err)
+		}
 	}

 	versionBucketPath := filepath.Join(p.versioningDir, bucket)
@@ -1303,10 +1388,12 @@ func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Accoun
 		originalMTime = srcInfo.ModTime()
 	}

-	// Prioritize copy_file_range for internal file-to-file version copies.
-	_, err = io.Copy(f.File(), sf)
-	if err != nil {
-		return versionPath, err
+	if !isDir {
+		// Prioritize copy_file_range for internal file-to-file version copies.
+		_, err = io.Copy(f.File(), sf)
+		if err != nil {
+			return versionPath, err
+		}
 	}

 	// Restore original mtime after copy
@@ -1325,6 +1412,9 @@ func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Accoun
 	// Copy the object attributes(metadata)
 	for _, attr := range attrs {
 		data, err := p.meta.RetrieveAttribute(sf, bucket, key, attr)
+		if isDir && errors.Is(err, meta.ErrNoSuchKey) {
+			continue
+		}
 		if err != nil {
 			return versionPath, fmt.Errorf("list %v attribute: %w", attr, err)
 		}
@@ -1479,53 +1569,21 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 				Truncated:      true,
 			}, nil
 		}
+		// a directory is listed under the key with the trailing slash
+		key := path
 		if d.IsDir() {
-			// directory object only happens if directory empty
-			// check to see if this is a directory object by checking etag
-			etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, path, etagkey)
-			if errors.Is(err, meta.ErrNoSuchKey) || errors.Is(err, fs.ErrNotExist) {
-				return nil, backend.ErrSkipObj
-			}
-			if err != nil {
-				return nil, fmt.Errorf("get etag: %w", err)
-			}
-			etag := string(etagBytes)
-
-			fi, err := d.Info()
-			if errors.Is(err, fs.ErrNotExist) {
-				return nil, backend.ErrSkipObj
-			}
-			if err != nil {
-				return nil, fmt.Errorf("get fileinfo: %w", err)
-			}
-
-			key := path + "/"
-			// Directory objects don't contain data
-			size := int64(0)
-			versionId := "null"
-
-			objects = append(objects, s3response.ObjectVersion{
-				ETag:         &etag,
-				Key:          &key,
-				LastModified: backend.GetTimePtr(fi.ModTime()),
-				IsLatest:     getBoolPtr(true),
-				Size:         &size,
-				VersionId:    &versionId,
-				StorageClass: types.ObjectVersionStorageClassStandard,
-			})
-
-			return &backend.ObjVersionFuncResult{
-				ObjectVersions: objects,
-				DelMarkers:     delMarkers,
-				Truncated:      availableObjCount == 1,
-			}, nil
+			key = path + "/"
 		}

-		// file object, get object info and fill out object data
+		// get object info and fill out object data
 		etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, path, etagkey)
 		if errors.Is(err, fs.ErrNotExist) {
 			return nil, backend.ErrSkipObj
 		}
+		if d.IsDir() && errors.Is(err, meta.ErrNoSuchKey) {
+			// a directory is listed only if it's a directory object
+			return nil, backend.ErrSkipObj
+		}
 		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
 			return nil, fmt.Errorf("get etag: %w", err)
 		}
@@ -1552,6 +1610,10 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 			}

 			size := fi.Size()
+			if d.IsDir() {
+				// directory objects don't contain data
+				size = 0
+			}

 			isDel, err := p.isObjDeleteMarker(bucket, path)
 			if err != nil {
@@ -1563,7 +1625,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 					IsLatest:     getBoolPtr(true),
 					VersionId:    &versionId,
 					LastModified: backend.GetTimePtr(fi.ModTime()),
-					Key:          &path,
+					Key:          &key,
 				})
 			} else {
 				// Retrieve checksum
@@ -1574,7 +1636,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {

 				objects = append(objects, s3response.ObjectVersion{
 					ETag:              &etag,
-					Key:               &path,
+					Key:               &key,
 					LastModified:      backend.GetTimePtr(fi.ModTime()),
 					Size:              &size,
 					VersionId:         &versionId,
@@ -1604,7 +1666,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		}

 		// List all the versions of the object in the versioning directory
-		versionPath := p.genObjVersionPath(bucket, path)
+		versionPath := p.genObjVersionPath(bucket, key)
 		dirEnts, err := os.ReadDir(versionPath)
 		if errors.Is(err, fs.ErrNotExist) {
 			return &backend.ObjVersionFuncResult{
@@ -1642,7 +1704,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 				nullObjDelMarker = &types.DeleteMarkerEntry{
 					VersionId:    backend.GetPtrFromString("null"),
 					LastModified: backend.GetTimePtr(nf.ModTime()),
-					Key:          &path,
+					Key:          &key,
 					IsLatest:     getBoolPtr(false),
 				}
 			} else {
@@ -1665,7 +1727,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {

 				nullVersionIdObj = &s3response.ObjectVersion{
 					ETag:         &etag,
-					Key:          &path,
+					Key:          &key,
 					LastModified: backend.GetTimePtr(nf.ModTime()),
 					Size:         &size,
 					VersionId:    backend.GetPtrFromString("null"),
@@ -1775,7 +1837,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 				delMarkers = append(delMarkers, types.DeleteMarkerEntry{
 					VersionId:    &versionId,
 					LastModified: backend.GetTimePtr(f.ModTime()),
-					Key:          &path,
+					Key:          &key,
 					IsLatest:     getBoolPtr(false),
 				})
 			} else {
@@ -1786,7 +1848,7 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 				}
 				objects = append(objects, s3response.ObjectVersion{
 					ETag:              &etag,
-					Key:               &path,
+					Key:               &key,
 					LastModified:      backend.GetTimePtr(f.ModTime()),
 					Size:              &size,
 					VersionId:         &versionId,
@@ -3889,8 +3951,23 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
 	if err != nil {
 		return s3response.CopyPartResult{}, fmt.Errorf("stat object: %w", err)
 	}
+	if strings.HasSuffix(srcObject, "/") != fi.IsDir() {
+		return s3response.CopyPartResult{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
+	}
+	// a directory object holds no data
+	srcSize := fi.Size()
+	if fi.IsDir() {
+		isObj, err := p.isDirObject(srcBucket, srcObject)
+		if err != nil {
+			return s3response.CopyPartResult{}, err
+		}
+		if !isObj {
+			return s3response.CopyPartResult{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
+		}
+		srcSize = 0
+	}

-	startOffset, length, err := backend.ParseCopySourceRange(fi.Size(), *upi.CopySourceRange)
+	startOffset, length, err := backend.ParseCopySourceRange(srcSize, *upi.CopySourceRange)
 	if err != nil {
 		return s3response.CopyPartResult{}, err
 	}
@@ -4128,11 +4205,22 @@ func (p *Posix) snapshotObjVersion(bucket, key string, vStatus types.BucketVersi
 		return nil
 	}

+	isDir := strings.HasSuffix(key, "/")
 	d, err := os.Stat(p.ObjectPath(bucket, key))
-	if err != nil || d.IsDir() {
+	if err != nil || d.IsDir() != isDir {
 		// nothing to snapshot
 		return nil
 	}
+	if isDir {
+		isObj, err := p.isDirObject(bucket, key)
+		if err != nil {
+			return err
+		}
+		if !isObj {
+			// nothing to snapshot
+			return nil
+		}
+	}

 	var isVersionIdMissing bool
 	if p.isBucketVersioningSuspended(vStatus) {
@@ -4268,6 +4356,37 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, err
 		}

+		expectedSum := getEmptyChecksumValue(checksumAlgorithm)
+		if checksumValue != "" && expectedSum != checksumValue {
+			return s3response.PutObjectOutput{}, s3err.GetChecksumBadDigestErr(checksumAlgorithm)
+		}
+
+		// reject object lock settings the bucket doesn't support before
+		// the directory object is changed
+		if po.ObjectLockLegalHoldStatus == types.ObjectLockLegalHoldStatusOn || po.ObjectLockMode != "" {
+			err = p.isBucketObjectLockEnabled(*po.Bucket)
+			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
+				return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
+			}
+			if err != nil {
+				return s3response.PutObjectOutput{}, err
+			}
+		}
+
+		vStatus, err := p.getBucketVersioningStatus(ctx, *po.Bucket)
+		if err != nil {
+			return s3response.PutObjectOutput{}, err
+		}
+		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.
+		err = p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
+		if err != nil {
+			return s3response.PutObjectOutput{}, err
+		}
+
 		err = p.mkdirAll(name, uid, gid, doChown)
 		if err != nil {
 			if errors.Is(err, syscall.EDQUOT) {
@@ -4279,6 +4398,23 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			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 {
+					return s3response.PutObjectOutput{}, err
+				}
+			} else {
+				versionID = ulid.Make().String()
+			}
+		}
+
 		err = p.storeObjectMetadata(nil, *po.Bucket, *po.Key, po.Metadata)
 		if err != nil {
 			return s3response.PutObjectOutput{}, fmt.Errorf("set object metadata: %w", err)
@@ -4292,8 +4428,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			}
 		}

-		expectedSum := getEmptyChecksumValue(checksumAlgorithm)
-
 		dirETag := emptyMD5
 		if p.dataIntegrityEtag {
 			dirETag = fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksumAlgorithm)), expectedSum)
@@ -4321,10 +4455,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			}
 		}

-		if checksumValue != "" && expectedSum != checksumValue {
-			return s3response.PutObjectOutput{}, s3err.GetChecksumBadDigestErr(checksumAlgorithm)
-		}
-
 		// set empty checksum
 		checksum := s3response.Checksum{
 			Type:      types.ChecksumTypeFullObject,
@@ -4338,9 +4468,29 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, fmt.Errorf("store checksum: %w", err)
 		}

-		// for directory object no version is created
+		if versionID != "" {
+			err = p.meta.StoreAttribute(nil, *po.Bucket, *po.Key, versionIdKey, []byte(versionID))
+			if err != nil {
+				return s3response.PutObjectOutput{}, fmt.Errorf("set versionId attr: %w", err)
+			}
+		}
+
+		if versioned {
+			// The directory mtime is the version's last modified time.
+			// Setting it needs the directory to be owned by the gateway,
+			// otherwise it's left as is.
+			now := time.Now()
+			_ = os.Chtimes(name, now, now)
+		}
+
+		err = p.putObjectLockSettings(ctx, po)
+		if err != nil {
+			return s3response.PutObjectOutput{}, err
+		}
+
 		return s3response.PutObjectOutput{
 			ETag:              dirETag,
+			VersionID:         versionID,
 			Size:              &contentLength,
 			ChecksumType:      checksum.Type,
 			ChecksumCRC32:     checksum.CRC32,
@@ -4614,34 +4764,9 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 		}
 	}

-	// Set object legal hold
-	if po.ObjectLockLegalHoldStatus == types.ObjectLockLegalHoldStatusOn {
-		err := p.PutObjectLegalHold(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", true)
-		if err != nil {
-			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
-				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
-			}
-			return s3response.PutObjectOutput{}, err
-		}
-	}
-
-	// Set object retention
-	if po.ObjectLockMode != "" {
-		retention := types.ObjectLockRetention{
-			Mode:            types.ObjectLockRetentionMode(po.ObjectLockMode),
-			RetainUntilDate: po.ObjectLockRetainUntilDate,
-		}
-		retParsed, err := json.Marshal(retention)
-		if err != nil {
-			return s3response.PutObjectOutput{}, fmt.Errorf("parse object lock retention: %w", err)
-		}
-		err = p.PutObjectRetention(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", retParsed)
-		if err != nil {
-			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
-				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
-			}
-			return s3response.PutObjectOutput{}, err
-		}
+	err = p.putObjectLockSettings(ctx, po)
+	if err != nil {
+		return s3response.PutObjectOutput{}, err
 	}

 	return s3response.PutObjectOutput{
@@ -4662,6 +4787,42 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 	}, nil
 }

+// putObjectLockSettings sets the legal hold and retention requested with
+// the put on the object that was just published
+func (p *Posix) putObjectLockSettings(ctx context.Context, po s3response.PutObjectInput) error {
+	// Set object legal hold
+	if po.ObjectLockLegalHoldStatus == types.ObjectLockLegalHoldStatusOn {
+		err := p.PutObjectLegalHold(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", true)
+		if err != nil {
+			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
+				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
+			}
+			return err
+		}
+	}
+
+	// Set object retention
+	if po.ObjectLockMode != "" {
+		retention := types.ObjectLockRetention{
+			Mode:            types.ObjectLockRetentionMode(po.ObjectLockMode),
+			RetainUntilDate: po.ObjectLockRetainUntilDate,
+		}
+		retParsed, err := json.Marshal(retention)
+		if err != nil {
+			return fmt.Errorf("parse object lock retention: %w", err)
+		}
+		err = p.PutObjectRetention(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", retParsed)
+		if err != nil {
+			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
+				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
+			}
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
 	release, err := p.acquireActionSlot(ctx)
 	if err != nil {
@@ -4721,8 +4882,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 			})
 	}

-	// Directory objects can't have versions
-	if !isDir && p.versioningEnabled() && vStatus != "" {
+	if p.versioningEnabled() && vStatus != "" {
 		if getString(input.VersionId) == "" {
 			// if the versionId is not specified, make the current version a delete marker
 			fi, err := os.Stat(objpath)
@@ -4736,6 +4896,16 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 			if err != nil {
 				return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 			}
+			if isDir {
+				isObj, err := p.isLiveDirObject(fi, bucket, object)
+				if err != nil {
+					return nil, err
+				}
+				if !isObj {
+					// AWS returns success if the object does not exist
+					return &s3.DeleteObjectOutput{}, nil
+				}
+			}

 			err = evalPreconditions(fi, bucket, object)
 			if err != nil {
@@ -4792,6 +4962,30 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 		} else {
 			versionPath := p.genObjVersionPath(bucket, object)

+			if isDir {
+				// the attributes at a directory object path may belong to a
+				// file or to a directory that isn't an object
+				fi, err := os.Stat(objpath)
+				if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
+					// AWS returns success if the object does not exist
+					return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
+				}
+				if isErrNameTooLong(err) {
+					return nil, s3err.GetKeyTooLongErr(int64(len(object)), 1024)
+				}
+				if err != nil {
+					return nil, fmt.Errorf("stat object: %w", err)
+				}
+				isObj, err := p.isLiveDirObject(fi, bucket, object)
+				if err != nil {
+					return nil, err
+				}
+				if !isObj {
+					// AWS returns success if the object does not exist
+					return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
+				}
+			}
+
 			vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
 			if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
 				// AWS returns success if the object does not exist
@@ -4829,6 +5023,16 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				if err != nil {
 					return nil, err
 				}
+				if isDir {
+					err = p.deleteDirObjectLatestVersion(bucket, object)
+					if err != nil {
+						return nil, err
+					}
+					return &s3.DeleteObjectOutput{
+						DeleteMarker: &isDelMarker,
+						VersionId:    input.VersionId,
+					}, nil
+				}
 				err = os.Remove(objpath)
 				if err != nil && !errors.Is(err, fs.ErrNotExist) && !isErrNotDir(err) {
 					return nil, fmt.Errorf("remove obj version: %w", err)
@@ -4862,7 +5066,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 					}, nil
 				}

-				srcObjVersion, err := ents[len(ents)-1].Info()
+				srcObjVersion, err := latestObjVersion(ents).Info()
 				if err != nil {
 					return nil, fmt.Errorf("get file info: %w", err)
 				}
@@ -5033,6 +5237,93 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 	return &s3.DeleteObjectOutput{}, nil
 }

+// latestObjVersion returns the entry of the version, among the version
+// directory entries, that becomes the latest one when the latest version
+// of the object is deleted
+func latestObjVersion(ents []fs.DirEntry) fs.DirEntry {
+	return ents[len(ents)-1]
+}
+
+// deleteDirObjectLatestVersion removes the latest version of the directory
+// object at bucket/key. The newest remaining version is restored onto the
+// directory. With no versions left the directory stops being an object and
+// is removed, unless other objects are under it.
+func (p *Posix) deleteDirObjectLatestVersion(bucket, key string) error {
+	objpath := p.ObjectPath(bucket, key)
+	versionPath := p.genObjVersionPath(bucket, key)
+
+	ents, err := os.ReadDir(versionPath)
+	if err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return fmt.Errorf("read version dir: %w", err)
+	}
+
+	if len(ents) == 0 {
+		err := os.Remove(objpath)
+		if isErrDirNotEmpty(err) {
+			// the directory stays as the parent of the objects under it
+			err = p.clearDirObjectAttrs(bucket, key)
+			if err != nil {
+				return err
+			}
+			err = p.meta.DeleteAttribute(bucket, key, etagkey)
+			if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
+				return fmt.Errorf("remove etag attribute: %w", err)
+			}
+			return nil
+		}
+		if err != nil && !errors.Is(err, fs.ErrNotExist) {
+			return fmt.Errorf("remove dir object: %w", err)
+		}
+		err = p.meta.DeleteAttributes(bucket, key)
+		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) && !errors.Is(err, fs.ErrNotExist) {
+			return fmt.Errorf("delete object attributes: %w", err)
+		}
+		p.removeParents(bucket, key)
+		return nil
+	}
+
+	srcVersion := latestObjVersion(ents)
+	srcVersionId := srcVersion.Name()
+	srcInfo, err := srcVersion.Info()
+	if err != nil {
+		return fmt.Errorf("get file info: %w", err)
+	}
+
+	// replace the attributes in place, so that the directory keeps its etag
+	for _, attr := range dirObjectAttrs {
+		data, err := p.meta.RetrieveAttribute(nil, versionPath, srcVersionId, attr)
+		if errors.Is(err, meta.ErrNoSuchKey) {
+			err = p.meta.DeleteAttribute(bucket, key, attr)
+			if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
+				return fmt.Errorf("remove %v attribute: %w", attr, err)
+			}
+			continue
+		}
+		if err != nil {
+			return fmt.Errorf("load %v attribute: %w", attr, err)
+		}
+
+		err = p.meta.StoreAttribute(nil, bucket, key, attr, data)
+		if err != nil {
+			return fmt.Errorf("store %v attribute: %w", attr, err)
+		}
+	}
+
+	err = os.Remove(filepath.Join(versionPath, srcVersionId))
+	if err != nil {
+		return fmt.Errorf("remove obj version: %w", err)
+	}
+
+	_ = p.meta.DeleteAttributes(versionPath, srcVersionId)
+	p.removeParents(filepath.Join(p.versioningDir, bucket), filepath.Join(genObjVersionKey(key), srcVersionId))
+
+	// The restored version keeps its last modified time. Setting it needs
+	// the directory to be owned by the gateway, otherwise it's left as is.
+	_ = os.Chtimes(objpath, time.Now(), srcInfo.ModTime())
+
+	return nil
+}
+
 func (p *Posix) removeParents(bucket, object string) {
 	// this will remove all parent directories that were not
 	// specifically uploaded with a put object. we detect
@@ -5195,12 +5486,12 @@ func (p *Posix) GetObject(ctx context.Context, input *s3.GetObjectInput) (*s3.Ge
 		// in '/') have an etag attribute. Directories created incidentally on the
 		// filesystem or as parent directories during object upload should not be
 		// accessible via get-object.
-		_, derr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
-		if errors.Is(derr, meta.ErrNoSuchKey) || errors.Is(derr, fs.ErrNotExist) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
+		isObj, err := p.isDirObject(bucket, object)
+		if err != nil {
+			return nil, err
 		}
-		if derr != nil {
-			return nil, fmt.Errorf("get dir etag: %w", derr)
+		if !isObj {
+			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 		}
 	}

@@ -5241,6 +5532,18 @@ func (p *Posix) GetObject(ctx context.Context, input *s3.GetObjectInput) (*s3.Ge
 		return nil, err
 	}

+	// If versioning is configured get the object versionId
+	if p.versioningEnabled() && versionId == "" {
+		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
+		if errors.Is(err, meta.ErrNoSuchKey) {
+			versionId = nullVersionId
+		} else if err != nil {
+			return nil, err
+		}
+
+		versionId = string(vId)
+	}
+
 	if fid.IsDir() {
 		_, _, _, err := backend.ParseObjectRange(0, *input.Range)
 		if err != nil {
@@ -5299,18 +5602,6 @@ func (p *Posix) GetObject(ctx context.Context, input *s3.GetObjectInput) (*s3.Ge
 		}, nil
 	}

-	// If versioning is configured get the object versionId
-	if p.versioningEnabled() && versionId == "" {
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, meta.ErrNoSuchKey) {
-			versionId = nullVersionId
-		} else if err != nil {
-			return nil, err
-		}
-
-		versionId = string(vId)
-	}
-
 	// openForRead opens with FILE_SHARE_DELETE on Windows so that a concurrent
 	// DeleteObject can call os.Remove on this file while the GET response body
 	// is still being streamed. On POSIX, os.Open is sufficient.
@@ -5535,12 +5826,12 @@ func (p *Posix) HeadObject(ctx context.Context, input *s3.HeadObjectInput) (*s3.
 		// in '/') have an etag attribute. Directories created incidentally on the
 		// filesystem or as parent directories during object upload should not be
 		// accessible via head-object.
-		_, derr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
-		if errors.Is(derr, meta.ErrNoSuchKey) || errors.Is(derr, fs.ErrNotExist) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
+		isObj, err := p.isDirObject(bucket, object)
+		if err != nil {
+			return nil, err
 		}
-		if derr != nil {
-			return nil, fmt.Errorf("get dir etag: %w", derr)
+		if !isObj {
+			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 		}
 	}

@@ -5878,7 +6169,21 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 	if !strings.HasSuffix(srcObject, "/") && fi.IsDir() {
 		return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
-	if fi.Size() > p.copyObjectThreshold {
+	// a directory object holds no data
+	srcSize := fi.Size()
+	var srcBody io.Reader = f
+	if fi.IsDir() {
+		isObj, err := p.isDirObject(srcBucket, srcObject)
+		if err != nil {
+			return s3response.CopyObjectOutput{}, err
+		}
+		if !isObj {
+			return s3response.CopyObjectOutput{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
+		}
+		srcSize = 0
+		srcBody = strings.NewReader("")
+	}
+	if srcSize > p.copyObjectThreshold {
 		return s3response.CopyObjectOutput{}, s3err.GetCopySourceObjectTooLargeErr(p.copyObjectThreshold)
 	}

@@ -6042,7 +6347,7 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 			}
 		}
 	} else {
-		contentLength := fi.Size()
+		contentLength := srcSize

 		checksums, err := p.retrieveChecksums(f, srcBucket, srcObject)
 		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
@@ -6058,7 +6363,7 @@ func (p *Posix) CopyObject(ctx context.Context, input s3response.CopyObjectInput
 		putObjectInput := s3response.PutObjectInput{
 			Bucket:                    &dstBucket,
 			Key:                       &dstObject,
-			Body:                      f,
+			Body:                      srcBody,
 			ContentLength:             &contentLength,
 			ChecksumAlgorithm:         checksums.Algorithm,
 			ContentType:               input.ContentType,
@@ -6237,7 +6542,7 @@ func (p *Posix) FileToObj(bucket string, fetchOwner bool) backend.GetObjFunc {
 			}
 		}
 		if d.IsDir() {
-			// directory object only happens if directory empty
+			// a directory is listed only if it's a directory object
 			// check to see if this is a directory object by checking etag
 			etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, path, etagkey)
 			if errors.Is(err, meta.ErrNoSuchKey) || errors.Is(err, fs.ErrNotExist) {
@@ -6248,6 +6553,12 @@ func (p *Posix) FileToObj(bucket string, fetchOwner bool) backend.GetObjFunc {
 			}
 			etag := string(etagBytes)

+			// If the directory object is a delete marker, skip
+			isDel, _ := p.isObjDeleteMarker(bucket, path)
+			if isDel {
+				return s3response.Object{}, backend.ErrSkipObj
+			}
+
 			fi, err := d.Info()
 			if errors.Is(err, fs.ErrNotExist) {
 				return s3response.Object{}, backend.ErrSkipObj
@@ -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
 			}
@@ -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,
@@ -1941,6 +1941,7 @@ 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)
@@ -1957,6 +1958,7 @@ 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_delete_marker)
 	// GetObject action
@@ -1982,9 +1984,12 @@ 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_non_existing_object)
+	ts.Run(Versioning_DeleteObject_implicit_dir)
 	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)
@@ -1997,6 +2002,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)
@@ -2029,6 +2035,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)
@@ -3464,6 +3471,7 @@ 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,
@@ -3474,6 +3482,7 @@ func GetIntTests() IntTests {
 		"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_delete_marker":                                              Versioning_HeadObject_delete_marker,
 		"Versioning_GetObject_invalid_versionId":                                           Versioning_GetObject_invalid_versionId,
@@ -3495,8 +3504,11 @@ func GetIntTests() IntTests {
 		"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_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_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,
@@ -3507,6 +3519,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,
@@ -3534,6 +3547,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,
@@ -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,27 @@ func createObjVersions(client *s3.Client, bucket, object string, count int, opts
 	return versions, 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 {