Add multipart uploads cache for ListMultipartUploads() (#20407)

this cache will be honored only when `prefix=""` while
performing ListMultipartUploads() operation.

This is mainly to satisfy applications like alluxio
for their underfs implementation and tests.

replaces https://github.com/minio/minio/pull/20181

cmd/erasure-server-pool.go

@@ -45,6 +45,7 @@ import (
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/v3/sync/errgroup"
 	"github.com/minio/pkg/v3/wildcard"
+	"github.com/puzpuzpuz/xsync/v3"
 )
 
 type erasureServerPools struct {
@@ -63,6 +64,8 @@ type erasureServerPools struct {
 	decommissionCancelers []context.CancelFunc
 
 	s3Peer *S3PeerSys
+
+	mpCache *xsync.MapOf[string, MultipartInfo]
 }
 
 func (z *erasureServerPools) SinglePool() bool {
@@ -216,9 +219,37 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
 		break
 	}
 
+	// initialize the incomplete uploads cache
+	z.mpCache = xsync.NewMapOf[string, MultipartInfo]()
+
+	go z.cleanupStaleMPCache(ctx)
+
 	return z, nil
 }
 
+func (z *erasureServerPools) cleanupStaleMPCache(ctx context.Context) {
+	timer := time.NewTimer(globalAPIConfig.getStaleUploadsCleanupInterval())
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			z.mpCache.Range(func(id string, info MultipartInfo) bool {
+				if time.Since(info.Initiated) >= globalAPIConfig.getStaleUploadsExpiry() {
+					z.mpCache.Delete(id)
+					// No need to notify to peers, each node will delete its own cache.
+				}
+				return true
+			})
+
+			// Reset for the next interval
+			timer.Reset(globalAPIConfig.getStaleUploadsCleanupInterval())
+		}
+	}
+}
+
 func (z *erasureServerPools) NewNSLock(bucket string, objects ...string) RWLocker {
 	return z.serverPools[0].NewNSLock(bucket, objects...)
 }
@@ -1702,15 +1733,32 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 		return ListMultipartsInfo{}, err
 	}
 
-	if z.SinglePool() {
-		return z.serverPools[0].ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
-	}
-
 	poolResult := ListMultipartsInfo{}
 	poolResult.MaxUploads = maxUploads
 	poolResult.KeyMarker = keyMarker
 	poolResult.Prefix = prefix
 	poolResult.Delimiter = delimiter
+
+	// if no prefix provided, return the list from cache
+	if prefix == "" {
+		if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
+			return ListMultipartsInfo{}, toObjectErr(err, bucket)
+		}
+
+		z.mpCache.Range(func(_ string, mp MultipartInfo) bool {
+			poolResult.Uploads = append(poolResult.Uploads, mp)
+			return true
+		})
+		sort.Slice(poolResult.Uploads, func(i int, j int) bool {
+			return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated)
+		})
+		return poolResult, nil
+	}
+
+	if z.SinglePool() {
+		return z.serverPools[0].ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+	}
+
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
@@ -1722,15 +1770,27 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 		}
 		poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
 	}
+
 	return poolResult, nil
 }
 
 // Initiate a new multipart upload on a hashedSet based on object name.
-func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (*NewMultipartUploadResult, error) {
+func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (mp *NewMultipartUploadResult, err error) {
 	if err := checkNewMultipartArgs(ctx, bucket, object); err != nil {
 		return nil, err
 	}
 
+	defer func() {
+		if err == nil && mp != nil {
+			z.mpCache.Store(mp.UploadID, MultipartInfo{
+				Bucket:    bucket,
+				Object:    object,
+				UploadID:  mp.UploadID,
+				Initiated: time.Now(),
+			})
+		}
+	}()
+
 	if z.SinglePool() {
 		return z.serverPools[0].NewMultipartUpload(ctx, bucket, object, opts)
 	}
@@ -1874,11 +1934,18 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object
 }
 
 // Aborts an in-progress multipart operation on hashedSet based on the object name.
-func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error {
+func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) {
 	if err := checkAbortMultipartArgs(ctx, bucket, object, uploadID); err != nil {
 		return err
 	}
 
+	defer func() {
+		if err == nil {
+			z.mpCache.Delete(uploadID)
+			globalNotificationSys.DeleteUploadID(ctx, uploadID)
+		}
+	}()
+
 	if z.SinglePool() {
 		return z.serverPools[0].AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
 	}
@@ -1910,6 +1977,13 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		return objInfo, err
 	}
 
+	defer func() {
+		if err == nil {
+			z.mpCache.Delete(uploadID)
+			globalNotificationSys.DeleteUploadID(ctx, uploadID)
+		}
+	}()
+
 	if z.SinglePool() {
 		return z.serverPools[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
 	}
@@ -1952,6 +2026,12 @@ func (z *erasureServerPools) GetBucketInfo(ctx context.Context, bucket string, o
 	return bucketInfo, nil
 }
 
+// ClearUploadID deletes given uploadID from cache
+func (z *erasureServerPools) ClearUploadID(uploadID string) error {
+	z.mpCache.Delete(uploadID)
+	return nil
+}
+
 // DeleteBucket - deletes a bucket on all serverPools simultaneously,
 // even if one of the serverPools fail to delete buckets, we proceed to
 // undo a successful operation.

cmd/notification.go

@@ -691,6 +691,27 @@ func (sys *NotificationSys) ReloadPoolMeta(ctx context.Context) {
 	}
 }
 
+// DeleteUploadID notifies all the MinIO nodes to remove the
+// given uploadID from cache
+func (sys *NotificationSys) DeleteUploadID(ctx context.Context, uploadID string) {
+	ng := WithNPeers(len(sys.peerClients))
+	for idx, client := range sys.peerClients {
+		if client == nil {
+			continue
+		}
+		client := client
+		ng.Go(ctx, func() error {
+			return client.DeleteUploadID(ctx, uploadID)
+		}, idx, *client.host)
+	}
+	for _, nErr := range ng.Wait() {
+		reqInfo := (&logger.ReqInfo{}).AppendTags("peerAddress", nErr.Host.String())
+		if nErr.Err != nil {
+			peersLogOnceIf(logger.SetReqInfo(ctx, reqInfo), nErr.Err, nErr.Host.String())
+		}
+	}
+}
+
 // StopRebalance notifies all MinIO nodes to signal any ongoing rebalance
 // goroutine to stop.
 func (sys *NotificationSys) StopRebalance(ctx context.Context) {

cmd/peer-rest-client.go

@@ -467,6 +467,17 @@ func (client *peerRESTClient) ReloadPoolMeta(ctx context.Context) error {
 	return err
 }
 
+func (client *peerRESTClient) DeleteUploadID(ctx context.Context, uploadID string) error {
+	conn := client.gridConn()
+	if conn == nil {
+		return nil
+	}
+	_, err := cleanupUploadIDCacheMetaRPC.Call(ctx, conn, grid.NewMSSWith(map[string]string{
+		peerRESTUploadID: uploadID,
+	}))
+	return err
+}
+
 func (client *peerRESTClient) StopRebalance(ctx context.Context) error {
 	conn := client.gridConn()
 	if conn == nil {

cmd/peer-rest-common.go

@@ -67,6 +67,7 @@ const (
 	peerRESTStartRebalance  = "start-rebalance"
 	peerRESTMetrics         = "metrics"
 	peerRESTDryRun          = "dry-run"
+	peerRESTUploadID        = "up-id"
 
 	peerRESTURL         = "url"
 	peerRESTSha256Sum   = "sha256sum"

cmd/peer-rest-server.go

@@ -115,6 +115,7 @@ var (
 	signalServiceRPC               = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerSignalService, grid.NewMSS, grid.NewNoPayload)
 	stopRebalanceRPC               = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerStopRebalance, grid.NewMSS, grid.NewNoPayload)
 	updateMetacacheListingRPC      = grid.NewSingleHandler[*metacache, *metacache](grid.HandlerUpdateMetacacheListing, func() *metacache { return &metacache{} }, func() *metacache { return &metacache{} })
+	cleanupUploadIDCacheMetaRPC    = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerClearUploadID, grid.NewMSS, grid.NewNoPayload)
 
 	// STREAMS
 	// Set an output capacity of 100 for consoleLog and listenRPC
@@ -905,6 +906,26 @@ func (s *peerRESTServer) ReloadPoolMetaHandler(mss *grid.MSS) (np grid.NoPayload
 	return
 }
 
+func (s *peerRESTServer) HandlerClearUploadID(mss *grid.MSS) (np grid.NoPayload, nerr *grid.RemoteErr) {
+	objAPI := newObjectLayerFn()
+	if objAPI == nil {
+		return np, grid.NewRemoteErr(errServerNotInitialized)
+	}
+
+	pools, ok := objAPI.(*erasureServerPools)
+	if !ok {
+		return
+	}
+
+	// No need to return errors, this is not a highly strict operation.
+	uploadID := mss.Get(peerRESTUploadID)
+	if uploadID != "" {
+		pools.ClearUploadID(uploadID)
+	}
+
+	return
+}
+
 func (s *peerRESTServer) StopRebalanceHandler(mss *grid.MSS) (np grid.NoPayload, nerr *grid.RemoteErr) {
 	objAPI := newObjectLayerFn()
 	if objAPI == nil {