Revert "remove healObjects() from scanner"

This reverts commit 5258a68b456330e32d08290e21679cfc0ade5068.

cmd/data-crawler.go

@@ -578,6 +578,78 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 			return nil, err
 		}
 
+		if f.healObjectSelect == 0 {
+			// If we are not scanning, return now.
+			f.newCache.replaceHashed(thisHash, folder.parent, cache)
+			continue
+		}
+
+		objAPI := newObjectLayerFn()
+		if objAPI == nil {
+			continue
+		}
+
+		bgSeq, found := globalBackgroundHealState.getHealSequenceByToken(bgHealingUUID)
+		if !found {
+			continue
+		}
+
+		// Whatever remains in 'existing' are folders at this level
+		// that existed in the previous run but wasn't found now.
+		//
+		// This may be because of 2 reasons:
+		//
+		// 1) The folder/object was deleted.
+		// 2) We come from another disk and this disk missed the write.
+		//
+		// We therefore perform a heal check.
+		// If that doesn't bring it back we remove the folder and assume it was deleted.
+		// This means that the next run will not look for it.
+		for k := range existing {
+			bucket, prefix := path2BucketObject(k)
+			if f.dataUsageCrawlDebug {
+				logger.Info(color.Green("folder-scanner:")+" checking disappeared folder: %v/%v", bucket, prefix)
+			}
+
+			// Dynamic time delay.
+			t := UTCNow()
+
+			err = objAPI.HealObjects(ctx, bucket, prefix, madmin.HealOpts{
+				Recursive: true,
+				Remove:    healDeleteDangling,
+			},
+				func(bucket, object, versionID string) error {
+					// Wait for each heal as per crawler frequency.
+					sleepDuration(time.Since(t), f.dataUsageCrawlMult)
+
+					defer func() {
+						t = UTCNow()
+					}()
+					return bgSeq.queueHealTask(ctx,
+						healSource{
+							bucket:    bucket,
+							object:    object,
+							versionID: versionID,
+						}, madmin.HealItemObject)
+				})
+
+			sleepDuration(time.Since(t), f.dataUsageCrawlMult)
+
+			if f.dataUsageCrawlDebug && err != nil {
+				logger.Info(color.Green("healObjects:")+" checking returned value %v", err)
+			}
+
+			// Add unless healing returned an error.
+			if err == nil {
+				this := cachedFolder{name: k, parent: &thisHash, objectHealProbDiv: folder.objectHealProbDiv}
+				cache.addChild(hashPath(k))
+				if final {
+					f.existingFolders = append(f.existingFolders, this)
+				} else {
+					nextFolders = append(nextFolders, this)
+				}
+			}
+		}
 		f.newCache.replaceHashed(thisHash, folder.parent, cache)
 	}
 	return nextFolders, nil