fix: keep the object on DeleteObject with a versionId in a never-versioned bucket

Posix `DeleteObject` only compared `versionId` with the object's version when the bucket had a versioning status. A bucket that was never versioned has none, so the request fell through to the plain delete, which ignored `versionId`. Any well-formed `versionId` then permanently deleted the object, on a gateway with or without `--versioning-dir`, and through `DeleteObjects` as well, since it calls `DeleteObject` for each key.

A never-versioned bucket only holds `null` versions, so any other `versionId` names a version that doesn't exist. S3 returns success for that, echoes `x-amz-version-id`, and deletes nothing. `DeleteObject` now does the same, returning early for any `versionId` other than `null`. `versionId=null` still deletes the object.

backend/posix/posix.go

@@ -5284,6 +5284,13 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 		}
 	}
 
+	// a bucket that has never been versioned only holds null versions, so
+	// any other versionId names a version that doesn't exist: AWS returns
+	// success and deletes nothing
+	if versionId := getString(input.VersionId); versionId != "" && versionId != nullVersionId {
+		return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
+	}
+
 	fi, err := os.Stat(objpath)
 	if isErrNameTooLong(err) {
 		return nil, s3err.GetKeyTooLongErr(int64(len(object)), 1024)

tests/integration/group-tests.go

@@ -2023,8 +2023,10 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_DeleteObject_nested_dir_object)
 	ts.Run(Versioning_DeleteObject_non_existing_objects)
 	ts.Run(Versioning_DeleteObject_suspended)
+	ts.Run(Versioning_DeleteObject_never_versioned_bucket)
 	ts.Run(Versioning_DeleteObjects_success)
 	ts.Run(Versioning_DeleteObjects_delete_deleteMarkers)
+	ts.Run(Versioning_DeleteObjects_never_versioned_bucket)
 	// ListObjectVersions
 	ts.Run(ListObjectVersions_non_existing_bucket)
 	ts.Run(ListObjectVersions_negative_max_keys)
@@ -3570,8 +3572,10 @@ func GetIntTests() IntTests {
 		"Versioning_DeleteObject_nested_dir_object":                                        Versioning_DeleteObject_nested_dir_object,
 		"Versioning_DeleteObject_non_existing_objects":                                     Versioning_DeleteObject_non_existing_objects,
 		"Versioning_DeleteObject_suspended":                                                Versioning_DeleteObject_suspended,
+		"Versioning_DeleteObject_never_versioned_bucket":                                   Versioning_DeleteObject_never_versioned_bucket,
 		"Versioning_DeleteObjects_success":                                                 Versioning_DeleteObjects_success,
 		"Versioning_DeleteObjects_delete_deleteMarkers":                                    Versioning_DeleteObjects_delete_deleteMarkers,
+		"Versioning_DeleteObjects_never_versioned_bucket":                                  Versioning_DeleteObjects_never_versioned_bucket,
 		"ListObjectVersions_non_existing_bucket":                                           ListObjectVersions_non_existing_bucket,
 		"ListObjectVersions_negative_max_keys":                                             ListObjectVersions_negative_max_keys,
 		"ListObjectVersions_list_single_object_versions":                                   ListObjectVersions_list_single_object_versions,

tests/integration/versioning.go

@@ -2917,6 +2917,78 @@ func Versioning_DeleteObject_suspended(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_DeleteObject_never_versioned_bucket(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_never_versioned_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		versionId := "01KF2YVN948NAZ4JJR4X1AAVRA"
+		return forEachKey([]string{"my-obj", "my-dir/"}, func(obj string) error {
+			out, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+
+			// the object only has a null version, so deleting any other
+			// version succeeds without deleting the object
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &versionId,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+			if getString(res.VersionId) != versionId {
+				return fmt.Errorf("expected the versionId to be %v, instead got %v",
+					versionId, getString(res.VersionId))
+			}
+			if res.DeleteMarker != nil && *res.DeleteMarker {
+				return fmt.Errorf("expected the response DeleteMarker to be false")
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			head, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+			if getString(head.ETag) != getString(out.res.ETag) {
+				return fmt.Errorf("expected the ETag to be %v, instead got %v",
+					getString(out.res.ETag), getString(head.ETag))
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &nullVersionId,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err == nil {
+				return fmt.Errorf("expected NotFound, instead got nil")
+			}
+			return checkSdkApiErr(err, "NotFound")
+		})
+	})
+}
+
 func Versioning_DeleteObjects_success(s *S3Conf) error {
 	testName := "Versioning_DeleteObjects_success"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -3198,6 +3270,76 @@ func Versioning_DeleteObjects_delete_deleteMarkers(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_DeleteObjects_never_versioned_bucket(s *S3Conf) error {
+	testName := "Versioning_DeleteObjects_never_versioned_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		versionId := "01KF2YVN948NAZ4JJR4X1AAVRA"
+		objs := []string{"my-obj", "my-dir/"}
+		etags := map[string]string{}
+		objIds := []types.ObjectIdentifier{}
+		for _, obj := range objs {
+			out, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			etags[obj] = getString(out.res.ETag)
+			objIds = append(objIds, types.ObjectIdentifier{
+				Key:       &obj,
+				VersionId: &versionId,
+			})
+		}
+
+		// the objects only have a null version, so deleting any other
+		// version succeeds without deleting the objects
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: &bucket,
+			Delete: &types.Delete{
+				Objects: objIds,
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(res.Errors) != 0 {
+			return fmt.Errorf("expected no errors, instead got %v", res.Errors)
+		}
+		delResult := []types.DeletedObject{}
+		for _, objId := range objIds {
+			delResult = append(delResult, types.DeletedObject{
+				Key:       objId.Key,
+				VersionId: objId.VersionId,
+			})
+		}
+		if !compareDelObjects(delResult, res.Deleted) {
+			return fmt.Errorf("expected the deleted objects to be %v, instead got %v",
+				delResult, res.Deleted)
+		}
+
+		return forEachKey(objs, func(obj string) error {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			head, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+			if getString(head.ETag) != etags[obj] {
+				return fmt.Errorf("expected the ETag to be %v, instead got %v",
+					etags[obj], getString(head.ETag))
+			}
+			return nil
+		})
+	})
+}
+
 func Versioning_Multipart_Upload_success(s *S3Conf) error {
 	testName := "Versioning_Multipart_Upload_success"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {