perf(local): lease pooling and ReadDir pipelining for listing pipeline

Split stat_scheduler_core.go into core/lease/batch modules and replace
per-microbatch time.AfterFunc timers (688K goroutines/run) with a single
watchdog sweep over sync.Pool'd activeLease structs. Active lease set is
bounded by MaxWorkers (16), not microbatch count, eliminating ~3.44M heap
allocations per run.

Add prefetchReader that overlaps ReadDir(N+1) with stat processing of
batch N for multi-batch directories (>1024 entries). First batch is always
inlined — no goroutine overhead for the 90%+ of directories that are
single-batch. Uses cap-1 channel for handoff, sync.Once for fd ownership,
and a 5s close timeout to prevent indefinite hangs on stuck mounts.

Hardening: sweepLeases timeoutBuf accesses moved inside mu for race
safety, prefetchReader panic recovery now logs via fs.Errorf, and
includeDirFn errors logged at Infof instead of silently discarded.

Signed-off-by: Anagh Kumar Baranwal <6824881+darthShadow@users.noreply.github.com>

Author: darthShadow

backend/local/list.go

@@ -94,7 +94,7 @@ func (f *Fs) runTimedOperation(ctx context.Context, timer *time.Timer, operation
 	}
 }
 
-// retryTimedOperation reruns a timed operation when the timeout expires.
+// retryTimedOperation reruns operation indefinitely on DeadlineExceeded, backing off exponentially from 1s to 15 min, returning only on context cancellation or a non-timeout error.
 func (f *Fs) retryTimedOperation(ctx context.Context, name string, timer *time.Timer, operation func() error) error {
 	const maxBackoff = 15 * time.Minute
 	backoff := 1 * time.Second
@@ -202,11 +202,14 @@ func (f *Fs) listFileInfos(ctx context.Context, fd *os.File, openDir func() (*os
 	return allFis, nil
 }
 
-// listCachedFileInfos reads one directory sequentially. Each iteration reads one
-// ReadDir(1024) batch, filters it, stats the filtered entries through the
-// shared per-Fs scheduler, materializes the results, then advances to the next
-// ReadDir batch.
+// listCachedFileInfos reads one directory. The first ReadDir(1024) batch stays
+// inline so single-batch directories avoid prefetch overhead; if that batch
+// returns exactly readDirBatchSize entries (indicating more may follow), later
+// batches are read through prefetchReader so ReadDir for batch N+1 overlaps
+// stat work for batch N.
 func (f *Fs) listCachedFileInfos(ctx context.Context, fd *os.File, openDir func() (*os.File, error), preFilter func(os.DirEntry) (cachedDirEntry, bool), statFunc func(entry *cachedDirEntry) (os.FileInfo, error)) (allFis []statFileInfo, err error) {
+	const readDirBatchSize = 1024
+
 	defer func() {
 		if r := recover(); r != nil {
 			allFis = nil
@@ -286,37 +289,64 @@ func (f *Fs) listCachedFileInfos(ctx context.Context, fd *os.File, openDir func(
 	}
 
 	controller := newListController(f, f.statScheduler, listControllerOptions{})
-	defer controller.Close()
+	fdOwned := true
 	defer func() {
-		if fd != nil {
+		if fdOwned && fd != nil {
 			_ = fd.Close()
 		}
 	}()
 
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+
 	var entries []cachedDirEntry
-	for {
-		if err := ctx.Err(); err != nil {
-			return nil, err
+	batch := readResult{}
+	batch.entries, batch.err = fd.ReadDir(readDirBatchSize)
+	if len(batch.entries) > 0 {
+		// Give this ReadDir batch its own stat handle tied to the same opened
+		// directory so fstatat work can reuse one batch-scoped dirfd.
+		statDir, openErr := openDirAtReadFD(fd)
+		if openErr != nil {
+			fs.Debugf(f, "openDirAtReadFD: %v, falling back to entry.Info()", openErr)
+		} else {
+			batch.statDir = statDir
 		}
+	}
 
-		batch := readResult{}
-		batch.entries, batch.err = fd.ReadDir(1024)
-		if len(batch.entries) > 0 {
-			// Give this ReadDir batch its own stat handle tied to the same opened
-			// directory so fstatat work can reuse one batch-scoped dirfd.
-			statDir, openErr := openDirAtReadFD(fd)
-			if openErr != nil {
-				fs.Debugf(f, "openDirAtReadFD: %v, falling back to entry.Info()", openErr)
-			} else {
-				batch.statDir = statDir
+	var fis []statFileInfo
+	entries, fis, err = controller.ProcessBatch(ctx, &batch, entries, preFilter, statFunc)
+	if err != nil {
+		return nil, err
+	}
+	allFis = append(allFis, fis...)
+
+	if batch.err == io.EOF {
+		return allFis, nil
+	}
+	if batch.err != nil {
+		return nil, batch.err
+	}
+	if len(batch.entries) != readDirBatchSize {
+		return allFis, nil
+	}
+
+	reader := newPrefetchReader(ctx, f, fd)
+	fdOwned = false
+	defer reader.Close()
+
+	for {
+		batch, ok := reader.Next(ctx)
+		if !ok {
+			if err := ctx.Err(); err != nil {
+				return nil, err
 			}
+			return allFis, nil
 		}
 
-		var fis []statFileInfo
-		var statErr error
-		entries, fis, statErr = controller.ProcessBatch(ctx, &batch, entries, preFilter, statFunc)
-		if statErr != nil {
-			return nil, statErr
+		entries, fis, err = controller.ProcessBatch(ctx, &batch, entries, preFilter, statFunc)
+		if err != nil {
+			return nil, err
 		}
 		allFis = append(allFis, fis...)
 
@@ -402,7 +432,10 @@ func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err e
 	statFunc := func(entry *cachedDirEntry) (os.FileInfo, error) {
 		entryType := entry.Type()
 		if includeDirFn != nil && entryType.IsDir() {
-			include, _ := includeDirFn(entry.newRemote)
+			include, inclErr := includeDirFn(entry.newRemote)
+			if inclErr != nil {
+				fs.Infof(entry.newRemote, "directory exclusion check failed: %v", inclErr)
+			}
 			if !include {
 				return nil, nil
 			}
@@ -493,6 +526,6 @@ func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err e
 		}
 	}
 	// Uncomment to signal pre-filtered entries:
-	// ctx = filter.SetPreFiltered(ctx)
+	// FIXME: enable SetPreFiltered once preFilter in List correctly excludes all entries that IncludeRemote would reject
 	return entries, nil
 }

backend/local/list_pipeline_cleanup_test.go

@@ -74,7 +74,6 @@ func TestListController_ProcessBatch_ClosesStatDirOnPanic(t *testing.T) {
 		statDir: statDir,
 	}
 	controller := newListController(nil, nil, listControllerOptions{})
-	defer controller.Close()
 
 	var recovered any
 	func() {

backend/local/list_prefetch.go

@@ -0,0 +1,162 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/rclone/rclone/fs"
+)
+
+// prefetchReader overlaps ReadDir I/O for batch N+1 with foreground stat work
+// on batch N. It takes sole ownership of the read fd until Close returns.
+type prefetchReader struct {
+	ch     chan readResult
+	done   chan struct{}
+	cancel context.CancelFunc
+
+	closeFD sync.Once
+	fd      *os.File
+}
+
+// newPrefetchReader starts the background ReadDir loop and transfers sole
+// ownership of fd to the returned reader.
+func newPrefetchReader(ctx context.Context, owner *Fs, fd *os.File) *prefetchReader {
+	runCtx, cancel := context.WithCancel(ctx)
+	pr := &prefetchReader{
+		ch:     make(chan readResult, 1),
+		done:   make(chan struct{}),
+		cancel: cancel,
+		fd:     fd,
+	}
+	go pr.run(runCtx, owner, fd)
+	return pr
+}
+
+// run reads directory batches, opens an optional batch-scoped stat fd, and
+// hands completed readResult values to the foreground through a 1-deep buffer.
+func (pr *prefetchReader) run(ctx context.Context, owner *Fs, fd *os.File) {
+	var pending readResult
+
+	defer close(pr.done)
+	defer close(pr.ch)
+	defer pr.closeFD.Do(func() {
+		if pr.fd != nil {
+			_ = pr.fd.Close()
+		}
+	})
+	defer func() {
+		if r := recover(); r != nil {
+			closeReadResultStatDir(&pending)
+
+			var panicErr error
+			if e, ok := r.(error); ok {
+				panicErr = fmt.Errorf("panic reading directory: %w", e)
+			} else {
+				panicErr = fmt.Errorf("panic reading directory: %v", r)
+			}
+
+			fs.Errorf(owner, "prefetchReader: %v", panicErr)
+			select {
+			case pr.ch <- readResult{err: panicErr}:
+			case <-ctx.Done():
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		batch := readResult{}
+		batch.entries, batch.err = fd.ReadDir(1024)
+		pending = batch
+
+		if len(batch.entries) > 0 {
+			statDir, err := openDirAtReadFD(fd)
+			if err != nil {
+				fs.Debugf(owner, "openDirAtReadFD: %v, falling back to entry.Info()", err)
+			} else {
+				batch.statDir = statDir
+				pending.statDir = statDir
+			}
+		}
+
+		if ctx.Err() != nil {
+			closeReadResultStatDir(&batch)
+			return
+		}
+
+		select {
+		case pr.ch <- batch:
+			pending = readResult{}
+		case <-ctx.Done():
+			closeReadResultStatDir(&batch)
+			return
+		}
+
+		if batch.err != nil {
+			return
+		}
+	}
+}
+
+// Next blocks until the next prefetched batch is available or the caller
+// context is canceled.
+func (pr *prefetchReader) Next(ctx context.Context) (readResult, bool) {
+	select {
+	case batch, ok := <-pr.ch:
+		return batch, ok
+	case <-ctx.Done():
+		return readResult{}, false
+	}
+}
+
+// Close cancels the background reader, closes the read fd to interrupt any
+// blocked ReadDir, drains buffered batches, and waits for shutdown.
+func (pr *prefetchReader) Close() {
+	pr.cancel()
+	pr.closeFD.Do(func() {
+		if pr.fd != nil {
+			_ = pr.fd.Close()
+		}
+	})
+
+	timer := time.NewTimer(5 * time.Second)
+	defer timer.Stop()
+
+	for {
+		select {
+		case batch, ok := <-pr.ch:
+			if !ok {
+				select {
+				case <-pr.done:
+				case <-timer.C:
+					fs.Errorf(nil, "prefetchReader: background goroutine did not exit after close")
+				}
+				return
+			}
+			closeReadResultStatDir(&batch)
+		case <-pr.done:
+			for {
+				select {
+				case batch, ok := <-pr.ch:
+					if !ok {
+						return
+					}
+					closeReadResultStatDir(&batch)
+				default:
+					return
+				}
+			}
+		case <-timer.C:
+			fs.Errorf(nil, "prefetchReader: background goroutine did not exit after close")
+			return
+		}
+	}
+}