File locking module data store

private/bufpkg/bufmodule/bufmodulecache/bufmodulecache_test.go

@@ -16,9 +16,11 @@ package bufmodulecache
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmodulestore"
@@ -232,12 +234,16 @@ func TestConcurrentCacheReadWrite(t *testing.T) {
 		require.NoError(t, os.MkdirAll(cacheDir, 0755))
 		jobs, err := slicesext.MapError(
 			[]int{0, 1, 2, 3, 4},
-			func(_ int) (func(ctx context.Context) error, error) {
+			func(i int) (func(ctx context.Context) error, error) {
+				logger := logger.Named(fmt.Sprintf("job-%d", i))
 				bucket, err := storageos.NewProvider().NewReadWriteBucket(cacheDir)
 				if err != nil {
 					return nil, err
 				}
-				filelocker, err := filelock.NewLocker(cacheDir)
+				filelocker, err := filelock.NewLocker(
+					cacheDir,
+					filelock.LockerWithLockRetryDelay(10*time.Millisecond),
+				)
 				if err != nil {
 					return nil, err
 				}

private/bufpkg/bufmodule/bufmodulestore/module_data_store.go

@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
-	"os"
 
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule"
 	"github.com/bufbuild/buf/private/pkg/encoding"
@@ -345,35 +344,29 @@ func (p *moduleDataStore) deleteInvalidModuleData(
 			}
 		}()
 	}
-	moduleDir, err := os.Open(dirPath)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		// If the moduleDir was already closed, that is fine
-		if err := moduleDir.Close(); err != nil && err != fs.ErrClosed {
-			p.logger.Debug("failed_delete_close_module_dir", zap.Error(err))
-			retErr = multierr.Append(retErr, err)
+	var deleted map[string]struct{}
+	if err := p.bucket.Walk(ctx, dirPath, func(objectInfo storage.ObjectInfo) error {
+		components := normalpath.Components(objectInfo.Path())
+		if len(components) < 1 {
+			return fmt.Errorf("invalid object info path: %s", objectInfo.Path())
 		}
-	}()
-	fileNames, err := moduleDir.Readdirnames(-1)
-	if err != nil {
-		return err
-	}
-	// Close the moduleDir first before doing other operations
-	if err := moduleDir.Close(); err != nil {
-		return err
-	}
-	// Delete all contents except the lock file
-	for _, fileName := range fileNames {
-		if fileName != externalModuleDataLockFileName {
-			if err := p.bucket.Delete(ctx, dirPath+"/"+fileName); err != nil {
-				return err
-			}
+		topLevelPath := components[0]
+		if topLevelPath == externalModuleDataLockFileName {
+			return nil
 		}
+		if _, ok := deleted[topLevelPath]; ok {
+			return nil
+		}
+		if err := p.bucket.DeleteAll(ctx, topLevelPath); err != nil {
+			return err
+		}
+		deleted[topLevelPath] = struct{}{}
+		return nil
+	}); err != nil {
+		return err
 	}
-	// Delete the lock file
-	return p.bucket.Delete(ctx, dirPath+"/"+externalModuleDataLockFileName)
+	// Delete the entire dirPath
+	return p.bucket.DeleteAll(ctx, dirPath)
 }
 
 func (p *moduleDataStore) putModuleData(
@@ -517,7 +510,6 @@ func (p *moduleDataStore) putModuleData(
 		ctx,
 		filesBucket,
 		storage.MapWriteBucket(moduleCacheBucket, storage.MapOnPrefix(externalModuleDataFilesDir)),
-		storage.CopyWithAtomic(),
 	); err != nil {
 		return err
 	}

private/pkg/filelock/filelock.go

@@ -63,8 +63,32 @@ type Locker interface {
 //
 // The root directory path should generally be a data directory path.
 // The root directory must exist.
-func NewLocker(rootDirPath string) (Locker, error) {
-	return newLocker(rootDirPath)
+func NewLocker(rootDirPath string, options ...LockerOption) (Locker, error) {
+	return newLocker(rootDirPath, options...)
+}
+
+// LockerOption is an option for a new Locker.
+type LockerOption func(*lockerOptions)
+
+// LockerWithLockTimeout sets the default lock timeout for the Locker.
+//
+// If Lock/RLock is called with LockWithTimeout, that will override this default timeout.
+// If this is not set, the default lock timeout is 3s.
+func LockerWithLockTimeout(lockTimeout time.Duration) LockerOption {
+	return func(lockerOptions *lockerOptions) {
+		lockerOptions.lockTimeout = lockTimeout
+	}
+}
+
+// LockerWithLockRetryDelay sets the default lock retry delay for the Locker.
+//
+// If Lock/RLock is called with LockWithRetryDelay, that will override the default lock
+// retry delay.
+// If this is not set, the default lock retry delay is 200ms.
+func LockerWithLockRetryDelay(lockRetryDelay time.Duration) LockerOption {
+	return func(lockerOptions *lockerOptions) {
+		lockerOptions.lockRetryDelay = lockRetryDelay
+	}
 }
 
 // LockOption is an option for lock.

private/pkg/filelock/locker.go

@@ -18,15 +18,18 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/bufbuild/buf/private/pkg/normalpath"
 )
 
 type locker struct {
-	rootDirPath string
+	rootDirPath    string
+	lockTimeout    time.Duration
+	lockRetryDelay time.Duration
 }
 
-func newLocker(rootDirPath string) (*locker, error) {
+func newLocker(rootDirPath string, options ...LockerOption) (*locker, error) {
 	// allow symlinks
 	fileInfo, err := os.Stat(normalpath.Unnormalize(rootDirPath))
 	if err != nil {
@@ -35,16 +38,29 @@ func newLocker(rootDirPath string) (*locker, error) {
 	if !fileInfo.IsDir() {
 		return nil, fmt.Errorf("%q is not a directory", rootDirPath)
 	}
+	lockerOptions := newLockerOptions()
+	for _, option := range options {
+		option(lockerOptions)
+	}
 	return &locker{
 		// do not validate - allow anything including absolute paths and jumping context
-		rootDirPath: normalpath.Normalize(rootDirPath),
+		rootDirPath:    normalpath.Normalize(rootDirPath),
+		lockTimeout:    lockerOptions.lockTimeout,
+		lockRetryDelay: lockerOptions.lockRetryDelay,
 	}, nil
 }
 
 func (l *locker) Lock(ctx context.Context, path string, options ...LockOption) (Unlocker, error) {
 	if err := validatePath(path); err != nil {
 		return nil, err
 	}
+	options = append(
+		[]LockOption{
+			LockWithTimeout(l.lockTimeout),
+			LockWithRetryDelay(l.lockRetryDelay),
+		},
+		options..., // Any additional options set will be applied last
+	)
 	return lock(
 		ctx,
 		normalpath.Unnormalize(normalpath.Join(l.rootDirPath, path)),
@@ -56,10 +72,17 @@ func (l *locker) RLock(ctx context.Context, path string, options ...LockOption)
 	if err := validatePath(path); err != nil {
 		return nil, err
 	}
+	options = append(
+		[]LockOption{
+			LockWithTimeout(l.lockTimeout),
+			LockWithRetryDelay(l.lockRetryDelay),
+		},
+		options..., // Any additional options set will be applied last
+	)
 	return rlock(
 		ctx,
 		normalpath.Unnormalize(normalpath.Join(l.rootDirPath, path)),
-		options...,
+		options..., // Any additional options set will be applied last
 	)
 }
 
@@ -74,3 +97,15 @@ func validatePath(path string) error {
 	}
 	return nil
 }
+
+type lockerOptions struct {
+	lockTimeout    time.Duration
+	lockRetryDelay time.Duration
+}
+
+func newLockerOptions() *lockerOptions {
+	return &lockerOptions{
+		lockTimeout:    DefaultLockTimeout,
+		lockRetryDelay: DefaultLockRetryDelay,
+	}
+}