Merge pull request #2436 from versity/sis/null-version-position

fix: place the null version by its recorded position in posix versioning

backend/posix/posix.go

@@ -27,6 +27,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -191,6 +192,7 @@ const (
 	versioningKey       = "versioning"
 	deleteMarkerKey     = "delete-marker"
 	versionIdKey        = "version-id"
+	nullVersionPrevKey  = "null-version-prev"
 	partCrc64nvme       = "part-crc64nvme"
 	mpMetaKey           = "mp-metadata"
 
@@ -1520,6 +1522,21 @@ func (p *Posix) createObjVersion(bucket, key string, size int64, acc auth.Accoun
 		}
 	}
 
+	if versionId == nullVersionId {
+		// the null version id doesn't sort with the ulid version ids, so
+		// record its place in the version history: it's the current
+		// version, which makes it newer than the versions already in the
+		// versioning directory and older than any version created later
+		prev, err := newestObjVersionId(filepath.Join(versionBucketPath, genObjVersionKey(key)))
+		if err != nil {
+			return versionPath, fmt.Errorf("get newest object version: %w", err)
+		}
+		err = p.meta.StoreAttribute(f.File(), versionPath, "", nullVersionPrevKey, []byte(prev))
+		if err != nil {
+			return versionPath, fmt.Errorf("store %v attribute: %w", nullVersionPrevKey, err)
+		}
+	}
+
 	if err := f.link(); err != nil {
 		return versionPath, err
 	}
@@ -1702,10 +1719,11 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		if err == nil {
 			versionId = string(versionIdBytes)
 		}
-		if versionId == versionIdMarker {
-			*pastVersionIdMarker = true
-		}
-		if *pastVersionIdMarker {
+		// the version id marker is the version listed last, so the
+		// listing continues with the version following it
+		if !*pastVersionIdMarker {
+			*pastVersionIdMarker = versionId == versionIdMarker
+		} else {
 			fi, err := d.Info()
 			if errors.Is(err, fs.ErrNotExist) {
 				return nil, backend.ErrSkipObj
@@ -1794,11 +1812,17 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		// before starting the object versions listing
 		var nullVersionIdObj *s3response.ObjectVersion
 		var nullObjDelMarker *types.DeleteMarkerEntry
+		var nullPos nullVersionPos
 		nf, err := os.Stat(filepath.Join(versionPath, nullVersionId))
 		if err != nil && !errors.Is(err, fs.ErrNotExist) {
 			return nil, err
 		}
 		if err == nil {
+			nullPos, err = p.getNullVersionPos(versionPath, nf)
+			if err != nil {
+				return nil, err
+			}
+
 			isDel, err := p.isObjDeleteMarker(versionPath, nullVersionId)
 			if err != nil {
 				return nil, err
@@ -1847,34 +1871,39 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		}
 
 		isNullVersionIdObjFound := nullVersionIdObj != nil || nullObjDelMarker != nil
+		isNullVersionIdObjAdded := false
 
-		if len(dirEnts) == 1 && (isNullVersionIdObjFound) {
-			if nullObjDelMarker != nil {
-				delMarkers = append(delMarkers, *nullObjDelMarker)
+		// addNullVersion lists the null version at its place in the
+		// version history, or continues the listing after it when it's
+		// the version id marker. It returns the truncated result when
+		// the null version fills up the listing.
+		addNullVersion := func() *backend.ObjVersionFuncResult {
+			isNullVersionIdObjAdded = true
+			if !*pastVersionIdMarker {
+				*pastVersionIdMarker = versionIdMarker == nullVersionId
+				return nil
 			}
+
 			if nullVersionIdObj != nil {
 				objects = append(objects, *nullVersionIdObj)
 			}
+			if nullObjDelMarker != nil {
+				delMarkers = append(delMarkers, *nullObjDelMarker)
+			}
 
-			if availableObjCount == 1 {
+			if availableObjCount--; availableObjCount == 0 {
 				return &backend.ObjVersionFuncResult{
 					ObjectVersions:      objects,
 					DelMarkers:          delMarkers,
 					Truncated:           true,
 					NextVersionIdMarker: nullVersionId,
-				}, nil
-			} else {
-				return &backend.ObjVersionFuncResult{
-					ObjectVersions: objects,
-					DelMarkers:     delMarkers,
-				}, nil
+				}
 			}
+			return nil
 		}
 
-		isNullVersionIdObjAdded := false
+		for _, dEntry := range slices.Backward(dirEnts) {
 
-		for i := len(dirEnts) - 1; i >= 0; i-- {
-			dEntry := dirEnts[i]
 			// Skip the null versionId object to not
 			// break the object versions list
 			if dEntry.Name() == nullVersionId {
@@ -1890,26 +1919,11 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 			}
 
 			// If the null versionId object is found, first push it
-			// by checking its creation date, then continue the adding
-			if isNullVersionIdObjFound && !isNullVersionIdObjAdded {
-				if nf.ModTime().After(f.ModTime()) {
-					if nullVersionIdObj != nil {
-						objects = append(objects, *nullVersionIdObj)
-					}
-					if nullObjDelMarker != nil {
-						delMarkers = append(delMarkers, *nullObjDelMarker)
-					}
-
-					isNullVersionIdObjAdded = true
-
-					if availableObjCount--; availableObjCount == 0 {
-						return &backend.ObjVersionFuncResult{
-							ObjectVersions:      objects,
-							DelMarkers:          delMarkers,
-							Truncated:           true,
-							NextVersionIdMarker: nullVersionId,
-						}, nil
-					}
+			// by checking its place in the version history, then
+			// continue the adding
+			if isNullVersionIdObjFound && !isNullVersionIdObjAdded && nullPos.isNewerThan(f) {
+				if res := addNullVersion(); res != nil {
+					return res, nil
 				}
 			}
 			versionId := f.Name()
@@ -1979,20 +1993,8 @@ func (p *Posix) fileToObjVersions(bucket string) backend.GetVersionsFunc {
 		// If null versionId object is found but not yet pushed,
 		// push it after the listing, as it's the oldest object version
 		if isNullVersionIdObjFound && !isNullVersionIdObjAdded {
-			if nullVersionIdObj != nil {
-				objects = append(objects, *nullVersionIdObj)
-			}
-			if nullObjDelMarker != nil {
-				delMarkers = append(delMarkers, *nullObjDelMarker)
-			}
-
-			if availableObjCount--; availableObjCount == 0 {
-				return &backend.ObjVersionFuncResult{
-					ObjectVersions:      objects,
-					DelMarkers:          delMarkers,
-					Truncated:           true,
-					NextVersionIdMarker: nullVersionId,
-				}, nil
+			if res := addNullVersion(); res != nil {
+				return res, nil
 			}
 		}
 
@@ -5179,7 +5181,7 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 					}, nil
 				}
 
-				srcObjVersion, err := latestObjVersion(ents)
+				srcObjVersion, err := p.latestObjVersion(versionPath, ents)
 				if err != nil {
 					return nil, fmt.Errorf("get latest obj version: %w", err)
 				}
@@ -5231,6 +5233,11 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				}
 
 				for _, attr := range attrs {
+					// the place of the null version only applies in
+					// the versioning directory
+					if attr == nullVersionPrevKey {
+						continue
+					}
 					data, err := p.meta.RetrieveAttribute(nil, versionPath, srcVersionId, attr)
 					if err != nil {
 						return nil, fmt.Errorf("load %v attribute", attr)
@@ -5357,13 +5364,65 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 	return &s3.DeleteObjectOutput{}, nil
 }
 
-// latestObjVersion returns the version, among the version directory entries,
-// that becomes the latest one when the latest version of the object is
-// deleted. The entries are named after the version id and os.ReadDir sorts
-// them by name, which puts the ulid version ids in creation order. The null
-// version id doesn't sort with them, so it's placed by its modification time,
-// which is the time the version was created.
-func latestObjVersion(ents []fs.DirEntry) (fs.FileInfo, error) {
+// newestObjVersionId returns the id of the newest version, other than the
+// null version, in the object versioning directory at versionPath, or an
+// empty string if it has none. The entries are named after the version id
+// and os.ReadDir sorts them by name, which puts the ulid version ids in
+// creation order.
+func newestObjVersionId(versionPath string) (string, error) {
+	ents, err := os.ReadDir(versionPath)
+	if errors.Is(err, fs.ErrNotExist) {
+		return "", nil
+	}
+	if err != nil {
+		return "", err
+	}
+	for _, ent := range slices.Backward(ents) {
+		if ent.Name() != nullVersionId {
+			return ent.Name(), nil
+		}
+	}
+	return "", nil
+}
+
+// nullVersionPos is the place of the null version in the version history of
+// an object, among the versions ordered by their ulid version ids
+type nullVersionPos struct {
+	// prev is the id of the newest version older than the null version,
+	// empty if the null version is the oldest one
+	prev string
+	// modTime places a null version that was moved to the versioning
+	// directory before its place was recorded
+	modTime *time.Time
+}
+
+// isNewerThan reports whether the null version is newer than the version
+// fi, an entry of the object versioning directory named after its version id
+func (n nullVersionPos) isNewerThan(fi fs.FileInfo) bool {
+	if n.modTime != nil {
+		return n.modTime.After(fi.ModTime())
+	}
+	return fi.Name() <= n.prev
+}
+
+// getNullVersionPos returns the place of the null version, nullInfo, in the
+// object versioning directory at versionPath
+func (p *Posix) getNullVersionPos(versionPath string, nullInfo fs.FileInfo) (nullVersionPos, error) {
+	prev, err := p.meta.RetrieveAttribute(nil, versionPath, nullVersionId, nullVersionPrevKey)
+	if errors.Is(err, meta.ErrNoSuchKey) {
+		modTime := nullInfo.ModTime()
+		return nullVersionPos{modTime: &modTime}, nil
+	}
+	if err != nil {
+		return nullVersionPos{}, fmt.Errorf("get %v attribute: %w", nullVersionPrevKey, err)
+	}
+	return nullVersionPos{prev: string(prev)}, nil
+}
+
+// latestObjVersion returns the version, among the entries of the object
+// versioning directory at versionPath, that becomes the latest one when the
+// latest version of the object is deleted
+func (p *Posix) latestObjVersion(versionPath string, ents []fs.DirEntry) (fs.FileInfo, error) {
 	var latest, nullEnt fs.DirEntry
 	for _, ent := range ents {
 		if ent.Name() == nullVersionId {
@@ -5391,7 +5450,11 @@ func latestObjVersion(ents []fs.DirEntry) (fs.FileInfo, error) {
 		return nil, err
 	}
 
-	if nullInfo.ModTime().After(latestInfo.ModTime()) {
+	pos, err := p.getNullVersionPos(versionPath, nullInfo)
+	if err != nil {
+		return nil, err
+	}
+	if pos.isNewerThan(latestInfo) {
 		return nullInfo, nil
 	}
 
@@ -5436,7 +5499,7 @@ func (p *Posix) deleteDirObjectLatestVersion(bucket, key string) error {
 		return nil
 	}
 
-	srcInfo, err := latestObjVersion(ents)
+	srcInfo, err := p.latestObjVersion(versionPath, ents)
 	if err != nil {
 		return fmt.Errorf("get latest obj version: %w", err)
 	}

tests/integration/ListObjectVersions.go

@@ -599,6 +599,136 @@ func ListObjectVersions_single_null_versionId_object(s *S3Conf) error {
 	})
 }
 
+func ListObjectVersions_paginate_null_version(s *S3Conf) error {
+	testName := "ListObjectVersions_paginate_null_version"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		// the null version sits in the middle of the version history:
+		// it's an object for "a-obj" and "c-dir/" and a delete marker
+		// for "b-obj"
+		objs := []string{"a-obj", "b-obj", "c-dir/"}
+		oldVersions := map[string][]types.ObjectVersion{}
+		for _, obj := range objs {
+			versions, err := createObjVersions(s3client, bucket, obj, 2)
+			if err != nil {
+				return err
+			}
+			versions[0].IsLatest = getBoolPtr(false)
+			oldVersions[obj] = versions
+		}
+
+		err := putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusSuspended)
+		if err != nil {
+			return err
+		}
+
+		nullVersions := map[string][]types.ObjectVersion{}
+		for _, obj := range []string{"a-obj", "c-dir/"} {
+			size := objDataLen(obj, 100)
+			out, err := putObjectWithData(size, &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			nullVersions[obj] = []types.ObjectVersion{
+				{
+					ETag:         out.res.ETag,
+					IsLatest:     getBoolPtr(false),
+					Key:          &obj,
+					Size:         &size,
+					VersionId:    &nullVersionId,
+					StorageClass: types.ObjectVersionStorageClassStandard,
+				},
+			}
+		}
+
+		delMarkerObj := "b-obj"
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &delMarkerObj,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+		delMarkers := []types.DeleteMarkerEntry{
+			{
+				Key:       &delMarkerObj,
+				VersionId: &nullVersionId,
+				IsLatest:  getBoolPtr(false),
+			},
+		}
+
+		err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
+		if err != nil {
+			return err
+		}
+
+		versions := []types.ObjectVersion{}
+		for _, obj := range objs {
+			newVersions, err := createObjVersions(s3client, bucket, obj, 2)
+			if err != nil {
+				return err
+			}
+			versions = append(versions, newVersions...)
+			versions = append(versions, nullVersions[obj]...)
+			versions = append(versions, oldVersions[obj]...)
+		}
+
+		// the pages end on each of the versions, the null ones included,
+		// and each version is listed once
+		total := len(versions) + len(delMarkers)
+		for _, maxKeys := range []int32{1, 2, 3, 4, 1000} {
+			var gotVersions []types.ObjectVersion
+			var gotDelMarkers []types.DeleteMarkerEntry
+			var keyMarker, versionIdMarker *string
+			for page := 0; ; page++ {
+				if page > total {
+					return fmt.Errorf("max-keys %v: expected the listing to end within %v pages",
+						maxKeys, total)
+				}
+
+				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("max-keys %v, page %v: %w", maxKeys, page, err)
+				}
+
+				if count := len(out.Versions) + len(out.DeleteMarkers); count > int(maxKeys) {
+					return fmt.Errorf("max-keys %v, page %v: expected at most %v entries, instead got %v",
+						maxKeys, page, maxKeys, count)
+				}
+				gotVersions = append(gotVersions, out.Versions...)
+				gotDelMarkers = append(gotDelMarkers, out.DeleteMarkers...)
+
+				if out.IsTruncated == nil || !*out.IsTruncated {
+					break
+				}
+				keyMarker, versionIdMarker = out.NextKeyMarker, out.NextVersionIdMarker
+			}
+
+			if !compareVersions(versions, gotVersions) {
+				return fmt.Errorf("max-keys %v: expected the listed object versions to be %v, instead got %v",
+					maxKeys, sprintVersions(versions), sprintVersions(gotVersions))
+			}
+			if !compareDelMarkers(delMarkers, gotDelMarkers) {
+				return fmt.Errorf("max-keys %v: expected the listed delete markers to be %v, instead got %v",
+					maxKeys, delMarkers, gotDelMarkers)
+			}
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func ListObjectVersions_checksum(s *S3Conf) error {
 	testName := "ListObjectVersions_checksum"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {

tests/integration/group-tests.go

@@ -2017,6 +2017,7 @@ func TestVersioning(ts *TestState) {
 	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_latest_version_null_version_order)
 	ts.Run(Versioning_DeleteObject_non_existing_object)
 	ts.Run(Versioning_DeleteObject_implicit_dir)
 	ts.Run(Versioning_DeleteObject_trailing_slash_counterpart)
@@ -2041,6 +2042,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(ListObjectVersions_multiple_object_versions_truncated)
 	ts.Run(ListObjectVersions_with_delete_markers)
 	ts.Run(ListObjectVersions_containing_null_versionId_obj)
+	ts.Run(ListObjectVersions_paginate_null_version)
 	ts.Run(ListObjectVersions_single_null_versionId_object)
 	ts.Run(ListObjectVersions_checksum)
 	// Multipart upload
@@ -3574,6 +3576,7 @@ func GetIntTests() IntTests {
 		"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_latest_version_null_version_order":                        Versioning_DeleteObject_latest_version_null_version_order,
 		"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,
@@ -3596,6 +3599,7 @@ func GetIntTests() IntTests {
 		"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,
+		"ListObjectVersions_paginate_null_version":                                         ListObjectVersions_paginate_null_version,
 		"ListObjectVersions_single_null_versionId_object":                                  ListObjectVersions_single_null_versionId_object,
 		"ListObjectVersions_checksum":                                                      ListObjectVersions_checksum,
 		"Versioning_Multipart_Upload_success":                                              Versioning_Multipart_Upload_success,

tests/integration/versioning.go

@@ -2290,6 +2290,147 @@ func Versioning_DeleteObject_latest_version_with_null_version(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_DeleteObject_latest_version_null_version_order(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_latest_version_null_version_order"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		// the null version is either an object put or a delete marker
+		// created while versioning is suspended
+		cases := []struct {
+			obj       string
+			delMarker bool
+		}{
+			{"my-obj", false},
+			{"my-dir/", false},
+			{"my-dm-obj", true},
+			{"my-dm-dir/", true},
+		}
+		for _, c := range cases {
+			obj := c.obj
+			// the versions are told apart by a metadata entry, as a
+			// directory object carries no data
+			put := func(marker string) (string, error) {
+				ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+				defer cancel()
+				out, err := s3client.PutObject(ctx, &s3.PutObjectInput{
+					Bucket:   &bucket,
+					Key:      &obj,
+					Metadata: map[string]string{"marker": marker},
+				})
+				if err != nil {
+					return "", err
+				}
+				return getString(out.VersionId), nil
+			}
+
+			// history is the version ids from the newest to the oldest,
+			// with the marker each one is put with
+			type version struct {
+				id, marker string
+			}
+			var history []version
+			for _, marker := range []string{"v1", "v2"} {
+				id, err := put(marker)
+				if err != nil {
+					return err
+				}
+				history = append([]version{{id, marker}}, history...)
+			}
+
+			err := putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusSuspended)
+			if err != nil {
+				return err
+			}
+
+			// the null version sits in the middle of the version history
+			if c.delMarker {
+				ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+				out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+					Bucket: &bucket,
+					Key:    &obj,
+				})
+				cancel()
+				if err != nil {
+					return err
+				}
+				if getString(out.VersionId) != nullVersionId {
+					return fmt.Errorf("%v: expected the delete marker versionId to be %v, instead got %v",
+						obj, nullVersionId, getString(out.VersionId))
+				}
+				history = append([]version{{nullVersionId, ""}}, history...)
+			} else {
+				if _, err := put("null"); err != nil {
+					return err
+				}
+				history = append([]version{{nullVersionId, "null"}}, history...)
+			}
+
+			err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
+			if err != nil {
+				return err
+			}
+
+			for _, marker := range []string{"v3", "v4"} {
+				id, err := put(marker)
+				if err != nil {
+					return err
+				}
+				history = append([]version{{id, marker}}, history...)
+			}
+
+			// deleting the latest version makes the version created right
+			// before it the latest one, whether it's the null version or not
+			for i, deleted := range history {
+				ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+				_, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+					Bucket:    &bucket,
+					Key:       &obj,
+					VersionId: &deleted.id,
+				})
+				cancel()
+				if err != nil {
+					return fmt.Errorf("%v: delete version %v: %w", obj, deleted.id, err)
+				}
+
+				ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+				res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+					Bucket: &bucket,
+					Key:    &obj,
+				})
+				cancel()
+
+				// the object is left with no current version when the
+				// null delete marker or none of the versions is latest
+				if i+1 == len(history) || history[i+1].marker == "" {
+					if err == nil {
+						return fmt.Errorf("%v: expected the object to have no current version after deleting %v, instead got version %v",
+							obj, deleted.id, getString(res.VersionId))
+					}
+					if err := checkSdkApiErr(err, "NotFound"); err != nil {
+						return fmt.Errorf("%v: after deleting %v: %w", obj, deleted.id, err)
+					}
+					continue
+				}
+				if err != nil {
+					return fmt.Errorf("%v: head object after deleting %v: %w", obj, deleted.id, err)
+				}
+
+				expected := history[i+1]
+				if getString(res.VersionId) != expected.id {
+					return fmt.Errorf("%v: expected the current versionId after deleting %v to be %v, instead got %v",
+						obj, deleted.id, expected.id, getString(res.VersionId))
+				}
+				expectedMeta := map[string]string{"marker": expected.marker}
+				if !areMapsSame(res.Metadata, expectedMeta) {
+					return fmt.Errorf("%v: expected the object metadata after deleting %v to be %v, instead got %v",
+						obj, deleted.id, expectedMeta, res.Metadata)
+				}
+			}
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func Versioning_DeleteObject_non_existing_object(s *S3Conf) error {
 	testName := "Versioning_DeleteObject_non_existing_object"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {