From 4643efe6be46bbd3065c7a5c4ebc0660fe2cba70 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Mon, 14 Aug 2023 12:28:13 -0700
Subject: [PATCH] fix: add deadline worker pattern for local disk removers
 (#17845)

---
 cmd/erasure-multipart.go | 50 ++++++++++++++++++++++++----------------
 cmd/erasure.go           | 12 ++++++----
 2 files changed, 38 insertions(+), 24 deletions(-)

diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go
index f9aa44554..ab46dae99 100644
--- a/cmd/erasure-multipart.go
+++ b/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)
-			if now.Sub(fi.ModTime) > expiry {
-				removeAll(pathJoin(diskPath, minioMetaMultipartBucket, uploadIDPath))
-			}
-			wait()
-			return nil
+			w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+			return w.Run(func() error {
+				wait := deletedCleanupSleeper.Timer(ctx)
+				if now.Sub(fi.ModTime) > expiry {
+					removeAll(pathJoin(diskPath, minioMetaMultipartBucket, uploadIDPath))
+				}
+				wait()
+				return nil
+			})
 		})
 		vi, err := disk.StatVol(ctx, pathJoin(minioMetaMultipartBucket, shaDir))
 		if err != nil {
 			return nil
 		}
-		wait := deletedCleanupSleeper.Timer(ctx)
-		if now.Sub(vi.Created) > expiry {
-			// We are not deleting shaDir recursively here, if shaDir is empty
-			// and its older then we can happily delete it.
-			Remove(pathJoin(diskPath, minioMetaMultipartBucket, shaDir))
-		}
-		wait()
-		return nil
+		w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+		return w.Run(func() error {
+			wait := deletedCleanupSleeper.Timer(ctx)
+			if now.Sub(vi.Created) > expiry {
+				// We are not deleting shaDir recursively here, if shaDir is empty
+				// and its older then we can happily delete it.
+				Remove(pathJoin(diskPath, minioMetaMultipartBucket, shaDir))
+			}
+			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)
-		if now.Sub(vi.Created) > expiry {
-			removeAll(pathJoin(diskPath, minioMetaTmpBucket, tmpDir))
-		}
-		wait()
-		return nil
+		w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+		return w.Run(func() error {
+			wait := deletedCleanupSleeper.Timer(ctx)
+			if now.Sub(vi.Created) > expiry {
+				removeAll(pathJoin(diskPath, minioMetaTmpBucket, tmpDir))
+			}
+			wait()
+			return nil
+		})
 	})
 }
 
diff --git a/cmd/erasure.go b/cmd/erasure.go
index 12b8e9474..01ed9d294 100644
--- a/cmd/erasure.go
+++ b/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)
-					removeAll(pathJoin(diskPath, minioMetaTmpDeletedBucket, ddir))
-					wait()
-					return nil
+					w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+					return w.Run(func() error {
+						wait := deletedCleanupSleeper.Timer(ctx)
+						removeAll(pathJoin(diskPath, minioMetaTmpDeletedBucket, ddir))
+						wait()
+						return nil
+					})
 				})
 			}(disk)
 		}