allow deletes on directory objects to perform permanent deletes (#17132)

cmd/bucket-handlers.go

@@ -540,6 +540,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			oss[index].SetTransitionState(goi.TransitionedObject)
 		}
 
+		// All deletes on directory objects needs to be for `nullVersionID`
+		if isDirObject(object.ObjectName) && object.VersionID == "" {
+			object.VersionID = nullVersionID
+		}
+
 		if replicateDeletes {
 			dsc = checkReplicateDelete(ctx, bucket, ObjectToDelete{
 				ObjectV: ObjectV{

cmd/bucket-handlers_test.go

@@ -747,8 +747,13 @@ func testAPIDeleteMultipleObjectsHandler(obj ObjectLayer, instanceType, bucketNa
 
 	deletedObjects := make([]DeletedObject, len(requestList[0].Objects))
 	for i := range requestList[0].Objects {
+		var vid string
+		if isDirObject(requestList[0].Objects[i].ObjectName) {
+			vid = nullVersionID
+		}
 		deletedObjects[i] = DeletedObject{
 			ObjectName: requestList[0].Objects[i].ObjectName,
+			VersionID:  vid,
 		}
 	}
 
@@ -758,9 +763,14 @@ func testAPIDeleteMultipleObjectsHandler(obj ObjectLayer, instanceType, bucketNa
 	successRequest1 := encodeResponse(requestList[1])
 
 	deletedObjects = make([]DeletedObject, len(requestList[1].Objects))
-	for i := range requestList[0].Objects {
+	for i := range requestList[1].Objects {
+		var vid string
+		if isDirObject(requestList[0].Objects[i].ObjectName) {
+			vid = nullVersionID
+		}
 		deletedObjects[i] = DeletedObject{
 			ObjectName: requestList[1].Objects[i].ObjectName,
+			VersionID:  vid,
 		}
 	}
 

cmd/erasure-server-pool.go

@@ -937,15 +937,7 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 		return ObjectInfo{}, err
 	}
 
-	origObject := object
 	object = encodeDirObject(object)
-	// Only directory objects skip creating new versions.
-	if object != origObject && isDirObject(object) && data.Size() == 0 {
-		// Treat all directory PUTs to behave as if they are performed
-		// on an unversioned bucket.
-		opts.Versioned = false
-		opts.VersionSuspended = false
-	}
 
 	if z.SinglePool() {
 		if !isMinioMetaBucketName(bucket) {

cmd/object-api-options.go

@@ -174,6 +174,11 @@ func delOpts(ctx context.Context, r *http.Request, bucket, object string) (opts
 	// benefits of replication, make sure to apply version suspension
 	// only at bucket level instead.
 	opts.VersionSuspended = globalBucketVersioningSys.Suspended(bucket)
+	// For directory objects, delete `null` version permanently.
+	if isDirObject(object) && opts.VersionID == "" {
+		opts.VersionID = nullVersionID
+	}
+
 	delMarker := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarker))
 	if delMarker != "" {
 		switch delMarker {
@@ -321,9 +326,16 @@ func putOpts(ctx context.Context, r *http.Request, bucket, object string, metada
 	if err != nil {
 		return opts, err
 	}
+
 	opts.VersionID = vid
 	opts.Versioned = versioned
 	opts.VersionSuspended = versionSuspended
+
+	// For directory objects skip creating new versions.
+	if isDirObject(object) && vid == "" {
+		opts.VersionID = nullVersionID
+	}
+
 	opts.MTime = mtime
 	opts.ReplicationSourceLegalholdTimestamp = lholdtimestmp
 	opts.ReplicationSourceRetentionTimestamp = retaintimestmp

cmd/utils.go

@@ -1038,6 +1038,9 @@ func decodeDirObject(object string) string {
 }
 
 func isDirObject(object string) bool {
+	if obj := encodeDirObject(object); obj != object {
+		object = obj
+	}
 	return HasSuffix(object, globalDirSuffix)
 }