feat: Implement listing version 3.0 (#12605)

Co-authored-by: Harshavardhana <harsha@minio.io>

cmd/metacache-server-pool.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"path"
 	pathutil "path"
 	"strings"
 	"sync"
@@ -57,7 +56,7 @@ func renameAllBucketMetacache(epPath string) error {
 // Required important fields are Bucket, Prefix, Separator.
 // Other important fields are Limit, Marker.
 // List ID always derived from the Marker.
-func (z *erasureServerPools) listPath(ctx context.Context, o listPathOptions) (entries metaCacheEntriesSorted, err error) {
+func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (entries metaCacheEntriesSorted, err error) {
 	if err := checkListObjsArgs(ctx, o.Bucket, o.Prefix, o.Marker, z); err != nil {
 		return entries, err
 	}
@@ -95,140 +94,217 @@ func (z *erasureServerPools) listPath(ctx context.Context, o listPathOptions) (e
 	}
 
 	// Decode and get the optional list id from the marker.
-	o.Marker, o.ID = parseMarker(o.Marker)
-	o.Create = o.ID == ""
-	if o.ID == "" {
-		o.ID = mustGetUUID()
-	}
+	o.parseMarker()
 	o.BaseDir = baseDirFromPrefix(o.Prefix)
-	if o.discardResult {
-		// Override for single object.
-		o.BaseDir = o.Prefix
+	o.Transient = o.Transient || isReservedOrInvalidBucket(o.Bucket, false)
+	if o.Transient {
+		o.Create = false
 	}
 
-	// For very small recursive listings, don't same cache.
-	// Attempts to avoid expensive listings to run for a long
-	// while when clients aren't interested in results.
-	// If the client DOES resume the listing a full cache
-	// will be generated due to the marker without ID and this check failing.
-	if o.Limit < 10 && o.Marker == "" && o.Create && o.Recursive {
-		o.discardResult = true
-		o.Transient = true
-	}
-
-	var cache metacache
+	// We have 2 cases:
+	// 1) Cold listing, just list.
+	// 2) Returning, but with no id. Start async listing.
+	// 3) Returning, with ID, stream from list.
+	//
 	// If we don't have a list id we must ask the server if it has a cache or create a new.
-	if o.Create {
-		o.CurrentCycle = intDataUpdateTracker.current()
-		o.OldestCycle = globalNotificationSys.findEarliestCleanBloomFilter(ctx, path.Join(o.Bucket, o.BaseDir))
-		var cache metacache
+	if o.ID != "" && !o.Transient {
+		// Create or ping with handout...
 		rpc := globalNotificationSys.restClientFromHash(o.Bucket)
-		if isReservedOrInvalidBucket(o.Bucket, false) {
-			rpc = nil
-			o.Transient = true
-		}
-		// Apply prefix filter if enabled.
-		o.SetFilter()
-		if rpc == nil || o.Transient {
-			// Local
-			cache = localMetacacheMgr.findCache(ctx, o)
+		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+		var c *metacache
+		if rpc == nil {
+			resp := localMetacacheMgr.getBucket(ctx, o.Bucket).findCache(*o)
+			c = &resp
 		} else {
-			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-			defer cancel()
-			c, err := rpc.GetMetacacheListing(ctx, o)
-			if err != nil {
-				if errors.Is(err, context.Canceled) {
-					// Context is canceled, return at once.
-					// request canceled, no entries to return
-					return entries, io.EOF
-				}
-				if !errors.Is(err, context.DeadlineExceeded) {
-					logger.LogIf(ctx, err)
-				}
-				o.Transient = true
-				cache = localMetacacheMgr.findCache(ctx, o)
+			c, err = rpc.GetMetacacheListing(ctx, *o)
+		}
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				// Context is canceled, return at once.
+				// request canceled, no entries to return
+				return entries, io.EOF
+			}
+			if !errors.Is(err, context.DeadlineExceeded) {
+				// TODO: Remove, not really informational.
+				logger.LogIf(ctx, err)
+				o.debugln("listPath: deadline exceeded")
+			}
+			o.Transient = true
+			o.Create = false
+			o.ID = mustGetUUID()
+		} else {
+			if c.fileNotFound {
+				// No cache found, no entries found.
+				return entries, io.EOF
+			}
+			if c.status == scanStateError || c.status == scanStateNone {
+				o.ID = ""
+				o.Create = false
+				o.debugln("scan status", c.status, " - waiting a roundtrip to create")
 			} else {
-				cache = *c
+				// Continue listing
+				o.ID = c.id
 			}
 		}
-		if cache.fileNotFound {
-			// No cache found, no entries found.
-			return entries, io.EOF
-		}
-		// Only create if we created a new.
-		o.Create = o.ID == cache.id
-		o.ID = cache.id
 	}
 
+	if o.ID != "" && !o.Transient {
+		// We have an existing list ID, continue streaming.
+		if o.Create {
+			o.debugln("Creating", o)
+			entries, err = z.listAndSave(ctx, o)
+			if err == nil || err == io.EOF {
+				return entries, err
+			}
+			entries.truncate(0)
+		} else {
+			if o.pool < len(z.serverPools) && o.set < len(z.serverPools[o.pool].sets) {
+				o.debugln("Resuming", o)
+				entries, err = z.serverPools[o.pool].sets[o.set].streamMetadataParts(ctx, *o)
+				if err == nil {
+					return entries, nil
+				}
+			} else {
+				err = fmt.Errorf("invalid pool/set")
+				o.pool, o.set = 0, 0
+			}
+		}
+		if IsErr(err, []error{
+			nil,
+			context.Canceled,
+			context.DeadlineExceeded,
+			// io.EOF is expected and should be returned but no need to log it.
+			io.EOF,
+		}...) {
+			// Expected good errors we don't need to return error.
+			return entries, err
+		}
+		entries.truncate(0)
+		go func() {
+			rpc := globalNotificationSys.restClientFromHash(o.Bucket)
+			ctx, cancel := context.WithTimeout(GlobalContext, 5*time.Second)
+			defer cancel()
+			if c, err := rpc.GetMetacacheListing(ctx, *o); err == nil {
+				c.error = "no longer used"
+				c.status = scanStateError
+			}
+		}()
+		o.ID = ""
+	}
+
+	// Do listing in-place.
+	// Create output for our results.
+	// Create filter for results.
+	o.debugln("Raw List", o)
+	filterCh := make(chan metaCacheEntry, o.Limit)
+	filteredResults := o.gatherResults(filterCh)
+	listCtx, cancelList := context.WithCancel(ctx)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	var listErr error
+
+	go func(o listPathOptions) {
+		defer wg.Done()
+		o.Limit = 0
+		listErr = z.listMerged(listCtx, o, filterCh)
+		o.debugln("listMerged returned with", listErr)
+	}(*o)
+
+	entries, err = filteredResults()
+	cancelList()
+	wg.Wait()
+	if listErr != nil && !errors.Is(listErr, context.Canceled) {
+		return entries, listErr
+	}
+	truncated := entries.len() > o.Limit || err == nil
+	entries.truncate(o.Limit)
+	if !o.Transient && truncated {
+		if o.ID == "" {
+			entries.listID = mustGetUUID()
+		} else {
+			entries.listID = o.ID
+		}
+	}
+	if !truncated {
+		return entries, io.EOF
+	}
+	return entries, nil
+}
+
+// listMerged will list across all sets and return a merged results stream.
+// The result channel is closed when no more results are expected.
+func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) error {
 	var mu sync.Mutex
 	var wg sync.WaitGroup
 	var errs []error
 	allAtEOF := true
+	var inputs []chan metaCacheEntry
 	mu.Lock()
 	// Ask all sets and merge entries.
+	listCtx, cancelList := context.WithCancel(ctx)
+	defer cancelList()
 	for _, pool := range z.serverPools {
 		for _, set := range pool.sets {
 			wg.Add(1)
+			results := make(chan metaCacheEntry, 100)
+			inputs = append(inputs, results)
 			go func(i int, set *erasureObjects) {
 				defer wg.Done()
-				e, err := set.listPath(ctx, o)
+				err := set.listPath(listCtx, o, results)
 				mu.Lock()
 				defer mu.Unlock()
 				if err == nil {
 					allAtEOF = false
 				}
 				errs[i] = err
-				entries.merge(e, -1)
-
-				// Resolve non-trivial conflicts
-				entries.deduplicate(func(existing, other *metaCacheEntry) (replace bool) {
-					// Pick object over directory
-					if existing.isDir() && !other.isDir() {
-						return true
-					}
-					if !existing.isDir() && other.isDir() {
-						return false
-					}
-					eFIV, err := existing.fileInfo(o.Bucket)
-					if err != nil {
-						return true
-					}
-					oFIV, err := other.fileInfo(o.Bucket)
-					if err != nil {
-						return false
-					}
-					// Replace if modtime is newer
-					if !oFIV.ModTime.Equal(eFIV.ModTime) {
-						return oFIV.ModTime.After(eFIV.ModTime)
-					}
-					// Use NumVersions as a final tiebreaker.
-					return oFIV.NumVersions > eFIV.NumVersions
-				})
-				if entries.len() > o.Limit {
-					allAtEOF = false
-					entries.truncate(o.Limit)
-				}
 			}(len(errs), set)
 			errs = append(errs, nil)
 		}
 	}
 	mu.Unlock()
+
+	// Gather results to a single channel.
+	err := mergeEntryChannels(ctx, inputs, results, func(existing, other *metaCacheEntry) (replace bool) {
+		// Pick object over directory
+		if existing.isDir() && !other.isDir() {
+			return true
+		}
+		if !existing.isDir() && other.isDir() {
+			return false
+		}
+
+		eFIV, err := existing.fileInfo(o.Bucket)
+		if err != nil {
+			return true
+		}
+		oFIV, err := other.fileInfo(o.Bucket)
+		if err != nil {
+			return false
+		}
+		// Replace if modtime is newer
+		if !oFIV.ModTime.Equal(eFIV.ModTime) {
+			return oFIV.ModTime.After(eFIV.ModTime)
+		}
+		// Use NumVersions as a final tiebreaker.
+		return oFIV.NumVersions > eFIV.NumVersions
+	})
+
+	cancelList()
 	wg.Wait()
+	if err != nil {
+		return err
+	}
+	if contextCanceled(ctx) {
+		return ctx.Err()
+	}
 
 	if isAllNotFound(errs) {
-		// All sets returned not found.
-		go func() {
-			// Update master cache with that information.
-			cache.status = scanStateSuccess
-			cache.fileNotFound = true
-			o.updateMetacacheListing(cache, globalNotificationSys.restClientFromHash(o.Bucket))
-		}()
-		// cache returned not found, entries truncated.
-		return entries, io.EOF
+		return nil
 	}
 
 	for _, err := range errs {
-		if err == nil {
+		if err == nil || contextCanceled(ctx) {
 			allAtEOF = false
 			continue
 		}
@@ -236,15 +312,66 @@ func (z *erasureServerPools) listPath(ctx context.Context, o listPathOptions) (e
 			continue
 		}
 		logger.LogIf(ctx, err)
-		return entries, err
+		return err
 	}
-	truncated := entries.len() > o.Limit || !allAtEOF
-	entries.truncate(o.Limit)
-	if !o.discardResult {
-		entries.listID = o.ID
+	if allAtEOF {
+		// TODO" Maybe, maybe not
+		return io.EOF
 	}
-	if !truncated {
-		return entries, io.EOF
-	}
-	return entries, nil
+	return nil
+}
+
+func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions) (entries metaCacheEntriesSorted, err error) {
+	// Use ID as the object name...
+	o.pool = z.getAvailablePoolIdx(ctx, minioMetaBucket, o.ID, 10<<20)
+	if o.pool < 0 {
+		// No space or similar, don't persist the listing.
+		o.pool = 0
+		o.Create = false
+		o.ID = ""
+		o.Transient = true
+		return entries, errDiskFull
+	}
+	o.set = z.serverPools[o.pool].getHashedSetIndex(o.ID)
+	saver := z.serverPools[o.pool].sets[o.set]
+
+	// Disconnect from call above, but cancel on exit.
+	listCtx, cancel := context.WithCancel(GlobalContext)
+	saveCh := make(chan metaCacheEntry, metacacheBlockSize)
+	inCh := make(chan metaCacheEntry, metacacheBlockSize)
+	outCh := make(chan metaCacheEntry, o.Limit)
+
+	filteredResults := o.gatherResults(outCh)
+
+	mc := o.newMetacache()
+	meta := metaCacheRPC{meta: &mc, cancel: cancel, rpc: globalNotificationSys.restClientFromHash(o.Bucket), o: *o}
+
+	// Save listing...
+	go func() {
+		if err := saver.saveMetaCacheStream(listCtx, &meta, saveCh); err != nil {
+			meta.setErr(err.Error())
+		}
+		cancel()
+	}()
+
+	// Do listing...
+	go func(o listPathOptions) {
+		err := z.listMerged(listCtx, o, inCh)
+		if err != nil {
+			meta.setErr(err.Error())
+		}
+		o.debugln("listAndSave: listing", o.ID, "finished with ", err)
+	}(*o)
+
+	// Write listing to results and saver.
+	go func() {
+		for entry := range inCh {
+			outCh <- entry
+			saveCh <- entry
+		}
+		close(outCh)
+		close(saveCh)
+	}()
+
+	return filteredResults()
 }