Merlin0220 · Nov 11, 2026 · Nov 12, 2026 · Nov 12, 2026 · Nov 12, 2026
@@ -149,11 +149,45 @@ type Config struct {
 	NBDPoolSize                 int               `env:"NBD_POOL_SIZE"                 envDefault:"64"`
 	Services                    []string          `env:"ORCHESTRATOR_SERVICES"         envDefault:"orchestrator"`
 	PersistentVolumeMounts      map[string]string `env:"PERSISTENT_VOLUME_MOUNTS"`
+	// VolumeReadDirConcurrency is how the parallelism of a volume directory
+	// listing's per-entry stat calls scales with the number of entries: the
+	// stats are serial below Threshold, then run on Base workers plus one
+	// more per Step additional entries, capped at Max. Every stat is a round
+	// trip on the volume's storage, so this bounds the listing's cost there.
+	VolumeReadDirConcurrency ReadDirConcurrencyConfig
 
 	// TLS is the mutual TLS settings of the shared port and the peer hop.
 	TLS TLSConfig
 }
 
+// ReadDirConcurrencyConfig holds the VOLUME_READ_DIR_CONCURRENCY_* settings
+// that scale volume directory listings with the entry count; see the
+// VolumeReadDirConcurrency field of Config.
+type ReadDirConcurrencyConfig struct {
+	Threshold int `env:"VOLUME_READ_DIR_CONCURRENCY_THRESHOLD" envDefault:"1000"`
+	Base      int `env:"VOLUME_READ_DIR_CONCURRENCY_BASE"       envDefault:"8"`
+	Step      int `env:"VOLUME_READ_DIR_CONCURRENCY_STEP"       envDefault:"5000"`
+	Max       int `env:"VOLUME_READ_DIR_CONCURRENCY_MAX"        envDefault:"16"`
+}
+
+// Validate reports scaling settings a listing could not run with.
+func (c ReadDirConcurrencyConfig) Validate() error {
+	if c.Threshold < 0 {
+		return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_THRESHOLD must be at least 0, got %d", c.Threshold)
+	}
+	if c.Base < 1 {
+		return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_BASE must be at least 1, got %d", c.Base)
+	}
+	if c.Step < 1 {
+		return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_STEP must be at least 1, got %d", c.Step)
+	}
+	if c.Max < c.Base {
+		return fmt.Errorf("VOLUME_READ_DIR_CONCURRENCY_MAX must be at least VOLUME_READ_DIR_CONCURRENCY_BASE, got %d < %d", c.Max, c.Base)
+	}
+
+	return nil
+}
+
 // AdditionalClickhouseEndpoints returns the non-blank entries from
 // CLICKHOUSE_CONNECTION_STRINGS that are *in addition to* the singular
 // CLICKHOUSE_CONNECTION_STRING. Order is preserved; first occurrence wins on
@@ -218,6 +252,10 @@ func Parse() (Config, error) {
 		return config, err
 	}
 
+	if err := config.VolumeReadDirConcurrency.Validate(); err != nil {
+		return config, err
+	}
+
 	if config.PersistentVolumeMounts != nil {
 		for name, path := range config.PersistentVolumeMounts {
 			path = filepath.Clean(path)

@@ -71,6 +71,49 @@ func TestParse(t *testing.T) {
 		require.NoError(t, err)
 		assert.True(t, config.DisableStartupReclaim)
 	})
+
+	t.Run("volume read dir concurrency gets defaults", func(t *testing.T) {
+		config, err := Parse()
+		require.NoError(t, err)
+
+		assert.Equal(t, ReadDirConcurrencyConfig{Threshold: 1000, Base: 8, Step: 5000, Max: 16}, config.VolumeReadDirConcurrency)
+	})
+
+	t.Run("volume read dir concurrency can be overridden", func(t *testing.T) {
+		t.Setenv("VOLUME_READ_DIR_CONCURRENCY_THRESHOLD", "100")
+		t.Setenv("VOLUME_READ_DIR_CONCURRENCY_BASE", "2")
+		t.Setenv("VOLUME_READ_DIR_CONCURRENCY_STEP", "50")
+		t.Setenv("VOLUME_READ_DIR_CONCURRENCY_MAX", "6")
+
+		config, err := Parse()
+		require.NoError(t, err)
+
+		assert.Equal(t, ReadDirConcurrencyConfig{Threshold: 100, Base: 2, Step: 50, Max: 6}, config.VolumeReadDirConcurrency)
+	})
+
+	t.Run("volume read dir concurrency rejects unusable scaling", func(t *testing.T) {
+		tests := []struct {
+			name string
+			envs map[string]string
+		}{
+			{"negative threshold", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_THRESHOLD": "-1"}},
+			{"zero base", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_BASE": "0"}},
+			{"zero step", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_STEP": "0"}},
+			{"max below base", map[string]string{"VOLUME_READ_DIR_CONCURRENCY_BASE": "4", "VOLUME_READ_DIR_CONCURRENCY_MAX": "3"}},
+		}
+
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				for key, value := range test.envs {
+					t.Setenv(key, value)
+				}
+
+				_, err := Parse()
+				require.Error(t, err)
+				assert.ErrorContains(t, err, "VOLUME_READ_DIR_CONCURRENCY")
+			})
+		}
+	})
 }
 
 func TestAdditionalClickhouseEndpoints(t *testing.T) {

@@ -28,7 +28,15 @@ func (b *Builder) Chroot(volumeType string, teamID, volumeID uuid.UUID) (*Chroot
 		return nil, err
 	}
 
-	fs, err := Chroot(fullPath, WithMetadata("volume-id", volumeID.String()))
+	fs, err := Chroot(fullPath,
+		WithMetadata("volume-id", volumeID.String()),
+		WithReadDirScaling(ReadDirScaling{
+			Threshold: b.config.VolumeReadDirConcurrency.Threshold,
+			Base:      b.config.VolumeReadDirConcurrency.Base,
+			Step:      b.config.VolumeReadDirConcurrency.Step,
+			Max:       b.config.VolumeReadDirConcurrency.Max,
+		}),
+	)
 	if err != nil {
 		return nil, err
 	}

@@ -45,6 +45,10 @@ type Chrooted struct {
 	// Waiters never initiate or retry that close.
 	closeDone chan struct{}
 	closeErr  error
+
+	// readDirScaling is how the parallelism of ReadDir's per-entry stat
+	// calls scales with the number of entries; see WithReadDirScaling.
+	readDirScaling ReadDirScaling
 }
 
 type Option func(*Chrooted)
@@ -59,6 +63,14 @@ func WithMetadata(key, value string) Option {
 	}
 }
 
+// WithReadDirScaling sets how the parallelism of ReadDir's per-entry stat
+// calls scales with the number of entries listed.
+func WithReadDirScaling(scaling ReadDirScaling) Option {
+	return func(fs *Chrooted) {
+		fs.readDirScaling = scaling
+	}
+}
+
 var (
 	// ErrClosed is returned by operations on a Chrooted after Close.
 	ErrClosed = errors.New("chrooted filesystem is closed")
@@ -124,9 +136,10 @@ func Chroot(source string, opts ...Option) (*Chrooted, error) {
 	}
 
 	fs := &Chrooted{
-		ActualRoot: source,
-		rootFD:     rootFD,
-		closeDone:  make(chan struct{}),
+		ActualRoot:     source,
+		rootFD:         rootFD,
+		closeDone:      make(chan struct{}),
+		readDirScaling: defaultReadDirScaling,
 	}
 
 	for _, opt := range opts {

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"syscall"
 
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/sys/unix"
 
 	"github.com/e2b-dev/infra/packages/shared/pkg/filesystem"
@@ -408,9 +409,49 @@ func (fs *Chrooted) TempFile(dir, prefix string) (*os.File, error) {
 	return nil, &os.PathError{Op: "createtemp", Path: path.Join(dir, prefix+"*"), Err: os.ErrExist}
 }
 
+// ReadDirScaling says how many per-entry stat calls ReadDir issues in
+// parallel as the listed directory grows: the stats are serial below
+// Threshold, then run on Base workers plus one more per Step additional
+// entries, capped at Max. Each stat is a single fstatat through the listing's
+// directory descriptor, so the calls are independent of each other and safe
+// to run concurrently. Workers is total: the zero value and degenerate
+// fields yield 1, never a panic.
+type ReadDirScaling struct {
+	Threshold int
+	Base      int
+	Step      int
+	Max       int
+}
+
+// Workers returns how many of a listing's entries may be statted in parallel.
+func (s ReadDirScaling) Workers(entries int) int {
+	if entries < s.Threshold {
+		return 1
+	}
+
+	workers := s.Base
+	if s.Step > 0 {
+		workers += (entries - s.Threshold) / s.Step
+	}
+	if s.Max > 0 && workers > s.Max {
+		workers = s.Max
+	}
+
+	return max(workers, 1)
+}
+
+// defaultReadDirScaling is the scaling a Chroot gets unless an option
+// overrides it; production values come from the orchestrator's config.
+var defaultReadDirScaling = ReadDirScaling{Threshold: 1000, Base: 8, Step: 5000, Max: 16}
+
 // ReadDir lists the directory at dirPath sorted by name, as os.ReadDir does.
 // Each entry carries its own attributes, not those of a symlink's target. An
 // entry removed while the directory is being read is left out.
+//
+// The per-entry stats run in parallel once the directory is large enough,
+// scaled by the fs's ReadDirScaling: the tree is usually an NFS volume, where
+// every stat is a network round trip. Smaller listings stay serial on the
+// calling goroutine — a worker pool would cost more than the stats.
 func (fs *Chrooted) ReadDir(dirPath string) ([]os.FileInfo, error) {
 	release, err := fs.acquire()
 	if err != nil {
@@ -432,21 +473,52 @@ func (fs *Chrooted) ReadDir(dirPath string) ([]os.FileInfo, error) {
 	}
 	slices.Sort(names)
 
-	infos := make([]os.FileInfo, 0, len(names))
-	for _, name := range names {
-		info, err := lstatAt(fd, name)
-		if err != nil {
-			if errors.Is(err, unix.ENOENT) {
-				continue
+	infos := make([]os.FileInfo, len(names))
+	statEntry := func(index int) error {
+		info, statErr := lstatAt(fd, names[index])
+		if statErr != nil {
+			if errors.Is(statErr, unix.ENOENT) {
+				// The entry was removed while the directory was being read.
+				return nil
 			}
 
-			return nil, pathError("readdir", path.Join(dirPath, name), err)
+			return pathError("readdir", path.Join(dirPath, names[index]), statErr)
 		}
 
-		infos = append(infos, info)
+		infos[index] = info
+
+		return nil
+	}
+
+	if workers := fs.readDirScaling.Workers(len(names)); workers > 1 {
+		group := new(errgroup.Group)
+		group.SetLimit(workers)
+
+		for index := range names {
+			group.Go(func() error {
+				return statEntry(index)
+			})
+		}
+
+		if err := group.Wait(); err != nil {
+			return nil, err
+		}
+	} else {
+		for index := range names {
+			if err := statEntry(index); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	listed := make([]os.FileInfo, 0, len(infos))
+	for _, info := range infos {
+		if info != nil {
+			listed = append(listed, info)
+		}
 	}
 
-	return infos, nil
+	return listed, nil
 }
 
 func (fs *Chrooted) Mkdir(filename string, perm os.FileMode) error {

@@ -0,0 +1,107 @@
+//go:build linux
+
+package chrooted
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReadDirScalingWorkers(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		scaling ReadDirScaling
+		entries int
+		want    int
+	}{
+		// The default scaling: serial below 1000 entries, 8 workers from
+		// there, one more per 5000 additional entries, capped at 16.
+		{"default: empty directory", defaultReadDirScaling, 0, 1},
+		{"default: below the threshold", defaultReadDirScaling, 999, 1},
+		{"default: at the threshold", defaultReadDirScaling, 1000, 8},
+		{"default: one below the first step", defaultReadDirScaling, 5999, 8},
+		{"default: at the first step", defaultReadDirScaling, 6000, 9},
+		{"default: at the second step", defaultReadDirScaling, 11000, 10},
+		{"default: one below the cap", defaultReadDirScaling, 40999, 15},
+		{"default: at the cap", defaultReadDirScaling, 41000, 16},
+		{"default: far above the cap", defaultReadDirScaling, 1000000, 16},
+
+		// The zero value is serial rather than explosive.
+		{"zero value", ReadDirScaling{}, 1000000, 1},
+
+		// A step of 0 adds no workers rather than dividing by zero.
+		{"zero step", ReadDirScaling{Threshold: 1, Base: 4, Step: 0, Max: 16}, 1000000, 4},
+
+		// A max below the base caps at the max.
+		{"max below the base", ReadDirScaling{Threshold: 1, Base: 8, Step: 1, Max: 2}, 1000000, 2},
+
+		// A base of 1 stays serial until a step adds workers.
+		{"base 1 grows with steps", ReadDirScaling{Threshold: 1, Base: 1, Step: 1, Max: 16}, 5, 5},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			assert.Equal(t, test.want, test.scaling.Workers(test.entries))
+		})
+	}
+}
+
+// TestReadDirScalingListing lists the same small directory once through the
+// default scaling (serial) and once through a scaling tiny enough to put the
+// per-entry stats on the worker pool, and expects the same sorted listing.
+func TestReadDirScalingListing(t *testing.T) {
+	t.Parallel()
+
+	source := t.TempDir()
+	const files = 40
+
+	want := make([]string, 0, files+1)
+	for i := range files {
+		name := fmt.Sprintf("file-%03d", i)
+		require.NoError(t, os.WriteFile(filepath.Join(source, name), []byte(name), 0o644))
+		want = append(want, name)
+	}
+	require.NoError(t, os.Mkdir(filepath.Join(source, "zz-dir"), 0o755))
+	want = append(want, "zz-dir")
+
+	serial, err := Chroot(source)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, serial.Close())
+	})
+
+	concurrent, err := Chroot(source,
+		WithReadDirScaling(ReadDirScaling{Threshold: 2, Base: 4, Step: 1, Max: 8}),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, concurrent.Close())
+	})
+
+	serialInfos, err := serial.ReadDir("/")
+	require.NoError(t, err)
+
+	concurrentInfos, err := concurrent.ReadDir("/")
+	require.NoError(t, err)
+
+	assert.Equal(t, want, entryNames(serialInfos))
+	assert.Equal(t, want, entryNames(concurrentInfos))
+}
+
+func entryNames(infos []os.FileInfo) []string {
+	names := make([]string, len(infos))
+	for i, info := range infos {
+		names[i] = info.Name()
+	}
+
+	return names
+}