lc: Apply DeleteAction correctly to objects (#11471)

When lifecycle decides to Delete an object and not a version in a
versioned bucket, the code should create a delete marker and not
removing the scanned version.

This commit fixes the issue.

cmd/bucket-lifecycle.go

@@ -65,13 +65,18 @@ func NewLifecycleSys() *LifecycleSys {
 	return &LifecycleSys{}
 }
 
+type expiryTask struct {
+	objInfo       ObjectInfo
+	versionExpiry bool
+}
+
 type expiryState struct {
-	expiryCh chan ObjectInfo
+	expiryCh chan expiryTask
 }
 
-func (es *expiryState) queueExpiryTask(oi ObjectInfo) {
+func (es *expiryState) queueExpiryTask(oi ObjectInfo, rmVersion bool) {
 	select {
-	case es.expiryCh <- oi:
+	case es.expiryCh <- expiryTask{objInfo: oi, versionExpiry: rmVersion}:
 	default:
 	}
 }
@@ -82,7 +87,7 @@ var (
 
 func newExpiryState() *expiryState {
 	es := &expiryState{
-		expiryCh: make(chan ObjectInfo, 10000),
+		expiryCh: make(chan expiryTask, 10000),
 	}
 	go func() {
 		<-GlobalContext.Done()
@@ -94,8 +99,8 @@ func newExpiryState() *expiryState {
 func initBackgroundExpiry(ctx context.Context, objectAPI ObjectLayer) {
 	globalExpiryState = newExpiryState()
 	go func() {
-		for oi := range globalExpiryState.expiryCh {
-			applyExpiryRule(ctx, objectAPI, oi, false)
+		for t := range globalExpiryState.expiryCh {
+			applyExpiryRule(ctx, objectAPI, t.objInfo, false, t.versionExpiry)
 		}
 	}()
 }

cmd/data-crawler.go

@@ -994,10 +994,12 @@ func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer,
 	return true
 }
 
-func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo) bool {
+func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, applyOnVersion bool) bool {
 	opts := ObjectOptions{}
 
-	opts.VersionID = obj.VersionID
+	if applyOnVersion {
+		opts.VersionID = obj.VersionID
+	}
 	if opts.VersionID == "" {
 		opts.Versioned = globalBucketVersioningSys.Enabled(obj.Bucket)
 	}
@@ -1029,20 +1031,20 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
 }
 
 // Apply object, object version, restored object or restored object version action on the given object
-func applyExpiryRule(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, restoredObject bool) bool {
+func applyExpiryRule(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, restoredObject, applyOnVersion bool) bool {
 	if obj.TransitionStatus != "" {
 		return applyExpiryOnTransitionedObject(ctx, objLayer, obj, restoredObject)
 	}
-	return applyExpiryOnNonTransitionedObjects(ctx, objLayer, obj)
+	return applyExpiryOnNonTransitionedObjects(ctx, objLayer, obj, applyOnVersion)
 }
 
 // Perform actions (removal of transitioning of objects), return true the action is successfully performed
 func applyLifecycleAction(ctx context.Context, action lifecycle.Action, objLayer ObjectLayer, obj ObjectInfo) (success bool) {
 	switch action {
 	case lifecycle.DeleteVersionAction, lifecycle.DeleteAction:
-		success = applyExpiryRule(ctx, objLayer, obj, false)
+		success = applyExpiryRule(ctx, objLayer, obj, false, action == lifecycle.DeleteVersionAction)
 	case lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction:
-		success = applyExpiryRule(ctx, objLayer, obj, true)
+		success = applyExpiryRule(ctx, objLayer, obj, true, action == lifecycle.DeleteRestoredVersionAction)
 	case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
 		success = applyTransitionAction(ctx, action, objLayer, obj)
 	}

cmd/object-handlers.go

@@ -433,7 +433,7 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
 	if lc, err := globalLifecycleSys.Get(bucket); err == nil {
 		action := evalActionFromLifecycle(ctx, *lc, objInfo, false)
 		if action == lifecycle.DeleteAction || action == lifecycle.DeleteVersionAction {
-			globalExpiryState.queueExpiryTask(objInfo)
+			globalExpiryState.queueExpiryTask(objInfo, action == lifecycle.DeleteVersionAction)
 			writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
 			return
 		}
@@ -600,7 +600,7 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
 	if lc, err := globalLifecycleSys.Get(bucket); err == nil {
 		action := evalActionFromLifecycle(ctx, *lc, objInfo, false)
 		if action == lifecycle.DeleteAction || action == lifecycle.DeleteVersionAction {
-			globalExpiryState.queueExpiryTask(objInfo)
+			globalExpiryState.queueExpiryTask(objInfo, action == lifecycle.DeleteVersionAction)
 			writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey))
 			return
 		}