File locking module data store

private/buf/bufcli/cache.go

@@ -29,6 +29,7 @@ import (
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmodulestore"
 	"github.com/bufbuild/buf/private/pkg/app/appext"
 	"github.com/bufbuild/buf/private/pkg/command"
+	"github.com/bufbuild/buf/private/pkg/filelock"
 	"github.com/bufbuild/buf/private/pkg/normalpath"
 	"github.com/bufbuild/buf/private/pkg/storage/storageos"
 )
@@ -96,6 +97,12 @@ var (
 	//
 	// Normalized.
 	v3CacheWKTRelDirPath = normalpath.Join("v3", "wellknowntypes")
+	// v3CacheModuleLockRelDirPath is the relative path to the lock files directory for module data. This path
+	// is relative to the files cache directory for module data and is used to store lock files for
+	// synchronizing reading and writing module data from the cache.
+	//
+	// Noramlized.
+	v3CacheModuleLockRelDirPath = "lock"
 )
 
 // NewModuleDataProvider returns a new ModuleDataProvider while creating the
@@ -166,12 +173,21 @@ func newModuleDataProvider(
 	if err != nil {
 		return nil, err
 	}
+	// Create filelocker directory and filelocker
+	if err := createCacheDir(container.CacheDirPath(), normalpath.Join(v3CacheModuleRelDirPath, v3CacheModuleLockRelDirPath)); err != nil {
+		return nil, err
+	}
+	filelocker, err := filelock.NewLocker(normalpath.Join(fullCacheDirPath, v3CacheModuleLockRelDirPath))
+	if err != nil {
+		return nil, err
+	}
 	return bufmodulecache.NewModuleDataProvider(
 		container.Logger(),
 		delegateModuleDataProvider,
 		bufmodulestore.NewModuleDataStore(
 			container.Logger(),
 			cacheBucket,
+			filelocker,
 		),
 	), nil
 }

private/bufpkg/bufmodule/bufmodulecache/bufmodulecache_test.go

@@ -16,15 +16,23 @@ 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"
 	"github.com/bufbuild/buf/private/bufpkg/bufmodule/bufmoduletesting"
+	"github.com/bufbuild/buf/private/pkg/filelock"
 	"github.com/bufbuild/buf/private/pkg/slicesext"
 	"github.com/bufbuild/buf/private/pkg/storage/storagemem"
+	"github.com/bufbuild/buf/private/pkg/storage/storageos"
+	"github.com/bufbuild/buf/private/pkg/thread"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
 )
 
 func TestCommitProviderForModuleKeyBasic(t *testing.T) {
@@ -165,6 +173,7 @@ func TestModuleDataProviderBasic(t *testing.T) {
 		bufmodulestore.NewModuleDataStore(
 			zap.NewNop(),
 			storagemem.NewReadWriteBucket(),
+			filelock.NewNopLocker(),
 		),
 	)
 
@@ -214,6 +223,64 @@ func TestModuleDataProviderBasic(t *testing.T) {
 	)
 }
 
+func TestConcurrentCacheReadWrite(t *testing.T) {
+	t.Parallel()
+
+	bsrProvider, moduleKeys := testGetBSRProviderAndModuleKeys(t, context.Background())
+	tempDir := t.TempDir()
+	cacheDir := filepath.Join(tempDir, "cache")
+	logger := zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel))
+
+	for i := 0; i < 20; i++ {
+		require.NoError(t, os.MkdirAll(cacheDir, 0755))
+		jobs, err := slicesext.MapError(
+			[]int{0, 1, 2, 3, 4},
+			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,
+					filelock.LockerWithLockRetryDelay(10*time.Millisecond), // Drops test time from ~16s to ~1s
+				)
+				if err != nil {
+					return nil, err
+				}
+				cacheProvider := newModuleDataProvider(
+					logger,
+					bsrProvider,
+					bufmodulestore.NewModuleDataStore(
+						logger,
+						bucket,
+						filelocker,
+					),
+				)
+				return func(ctx context.Context) error {
+					moduleDatas, err := cacheProvider.GetModuleDatasForModuleKeys(
+						ctx,
+						moduleKeys,
+					)
+					if err != nil {
+						return err
+					}
+					for _, moduleData := range moduleDatas {
+						// Calling moduleData.Bucket() checks the digest
+						if _, err := moduleData.Bucket(); err != nil {
+							return err
+						}
+					}
+					return nil
+				}, nil
+			},
+		)
+		require.NoError(t, err)
+		require.NoError(t, thread.Parallelize(context.Background(), jobs))
+		require.NoError(t, os.RemoveAll(cacheDir))
+	}
+}
+
 func testGetBSRProviderAndModuleKeys(t *testing.T, ctx context.Context) (bufmoduletesting.OmniProvider, []bufmodule.ModuleKey) {
 	bsrProvider, err := bufmoduletesting.NewOmniProvider(
 		bufmoduletesting.ModuleData{
@@ -235,7 +302,10 @@ func testGetBSRProviderAndModuleKeys(t *testing.T, ctx context.Context) (bufmodu
 		bufmoduletesting.ModuleData{
 			Name: "buf.build/foo/mod3",
 			PathToData: map[string][]byte{
-				"mod3.proto": []byte(
+				"mod3a.proto": []byte(
+					`syntax = proto3; package mod3;`,
+				),
+				"mod3b.proto": []byte(
 					`syntax = proto3; package mod3;`,
 				),
 			},