fix: add deadline worker pattern for local disk removers (#17845)

cmd/erasure-multipart.go

@@ -36,6 +36,7 @@ import (
 	"github.com/minio/minio/internal/crypto"
 	"github.com/minio/minio/internal/hash"
 	xhttp "github.com/minio/minio/internal/http"
+	xioutil "github.com/minio/minio/internal/ioutil"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/mimedb"
 	"github.com/minio/pkg/sync/errgroup"
@@ -205,25 +206,31 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
 			if err != nil {
 				return nil
 			}
-			wait := deletedCleanupSleeper.Timer(ctx)
+			w := xioutil.NewDeadlineWorker(diskMaxTimeout)
-			if now.Sub(fi.ModTime) > expiry {
+			return w.Run(func() error {
-				removeAll(pathJoin(diskPath, minioMetaMultipartBucket, uploadIDPath))
+				wait := deletedCleanupSleeper.Timer(ctx)
-			}
+				if now.Sub(fi.ModTime) > expiry {
-			wait()
+					removeAll(pathJoin(diskPath, minioMetaMultipartBucket, uploadIDPath))
-			return nil
+				}
+				wait()
+				return nil
+			})
 		})
 		vi, err := disk.StatVol(ctx, pathJoin(minioMetaMultipartBucket, shaDir))
 		if err != nil {
 			return nil
 		}
-		wait := deletedCleanupSleeper.Timer(ctx)
+		w := xioutil.NewDeadlineWorker(diskMaxTimeout)
-		if now.Sub(vi.Created) > expiry {
+		return w.Run(func() error {
-			// We are not deleting shaDir recursively here, if shaDir is empty
+			wait := deletedCleanupSleeper.Timer(ctx)
-			// and its older then we can happily delete it.
+			if now.Sub(vi.Created) > expiry {
-			Remove(pathJoin(diskPath, minioMetaMultipartBucket, shaDir))
+				// We are not deleting shaDir recursively here, if shaDir is empty
-		}
+				// and its older then we can happily delete it.
-		wait()
+				Remove(pathJoin(diskPath, minioMetaMultipartBucket, shaDir))
-		return nil
+			}
+			wait()
+			return nil
+		})
 	})
 
 	readDirFn(pathJoin(diskPath, minioMetaTmpBucket), func(tmpDir string, typ os.FileMode) error {
@@ -234,12 +241,15 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
 		if err != nil {
 			return nil
 		}
-		wait := deletedCleanupSleeper.Timer(ctx)
+		w := xioutil.NewDeadlineWorker(diskMaxTimeout)
-		if now.Sub(vi.Created) > expiry {
+		return w.Run(func() error {
-			removeAll(pathJoin(diskPath, minioMetaTmpBucket, tmpDir))
+			wait := deletedCleanupSleeper.Timer(ctx)
-		}
+			if now.Sub(vi.Created) > expiry {
-		wait()
+				removeAll(pathJoin(diskPath, minioMetaTmpBucket, tmpDir))
-		return nil
+			}
+			wait()
+			return nil
+		})
 	})
 }
 

cmd/erasure.go

@@ -31,6 +31,7 @@ import (
 	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/bpool"
 	"github.com/minio/minio/internal/dsync"
+	xioutil "github.com/minio/minio/internal/ioutil"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/sync/errgroup"
 )
@@ -342,10 +343,13 @@ func (er erasureObjects) cleanupDeletedObjects(ctx context.Context) {
 				defer wg.Done()
 				diskPath := disk.Endpoint().Path
 				readDirFn(pathJoin(diskPath, minioMetaTmpDeletedBucket), func(ddir string, typ os.FileMode) error {
-					wait := deletedCleanupSleeper.Timer(ctx)
+					w := xioutil.NewDeadlineWorker(diskMaxTimeout)
-					removeAll(pathJoin(diskPath, minioMetaTmpDeletedBucket, ddir))
+					return w.Run(func() error {
-					wait()
+						wait := deletedCleanupSleeper.Timer(ctx)
-					return nil
+						removeAll(pathJoin(diskPath, minioMetaTmpDeletedBucket, ddir))
+						wait()
+						return nil
+					})
 				})
 			}(disk)
 		}