Fix file locking for the ModuleDataStore (#3139)

Author: doriable

.dockerignore

@@ -14,6 +14,7 @@ private/buf/cmd/buf/command/alpha/protoc/internal/protoc-gen-insertion-point-rec
 private/buf/cmd/buf/command/alpha/protoc/internal/protoc-gen-insertion-point-writer/protoc-gen-insertion-point-writer
 private/buf/cmd/buf/command/alpha/protoc/test.txt
 private/buf/cmd/buf/command/generate/internal/protoc-gen-top-level-type-names-yaml/protoc-gen-top-level-type-names-yaml
+private/buf/cmd/buf/testdata/imports/*/v3/modulelocks/
 private/bufpkg/bufmodule/bufmoduleapi/cmd/buf-legacyfederation-go-data/buf-legacyfederation-go-data
 private/bufpkg/bufmodule/bufmoduletesting/cmd/buf-digest/buf-digest
 private/bufpkg/bufmodule/bufmoduletesting/cmd/buf-new-commit-id/buf-new-commit-id

.gitignore

@@ -14,6 +14,7 @@
 /private/buf/cmd/buf/command/alpha/protoc/internal/protoc-gen-insertion-point-writer/protoc-gen-insertion-point-writer
 /private/buf/cmd/buf/command/alpha/protoc/test.txt
 /private/buf/cmd/buf/command/generate/internal/protoc-gen-top-level-type-names-yaml/protoc-gen-top-level-type-names-yaml
+/private/buf/cmd/buf/testdata/imports/*/v3/modulelocks/
 /private/bufpkg/bufmodule/bufmoduleapi/cmd/buf-legacyfederation-go-data/buf-legacyfederation-go-data
 /private/bufpkg/bufmodule/bufmoduletesting/cmd/buf-digest/buf-digest
 /private/bufpkg/bufmodule/bufmoduletesting/cmd/buf-new-commit-id/buf-new-commit-id

make/buf/all.mk

@@ -31,7 +31,8 @@ FILE_IGNORES := $(FILE_IGNORES) \
 	private/buf/cmd/buf/command/alpha/protoc/test.txt \
 	private/bufpkg/buftesting/cache/ \
 	private/buf/buftesting/cache/ \
-	private/pkg/storage/storageos/tmp/
+	private/pkg/storage/storageos/tmp/ \
+	private/buf/cmd/buf/testdata/imports/*/v3/modulelocks/
 LICENSE_HEADER_LICENSE_TYPE := apache
 LICENSE_HEADER_COPYRIGHT_HOLDER := Buf Technologies, Inc.
 LICENSE_HEADER_YEAR_RANGE := 2020-2024

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"
 )
@@ -50,6 +51,7 @@ var (
 		v3CacheModuleRelDirPath,
 		v3CacheCommitsRelDirPath,
 		v3CacheWKTRelDirPath,
+		v3CacheModuleLockRelDirPath,
 	}
 
 	// v1CacheModuleDataRelDirPath is the relative path to the cache directory where module data
@@ -96,6 +98,11 @@ var (
 	//
 	// Normalized.
 	v3CacheWKTRelDirPath = normalpath.Join("v3", "wellknowntypes")
+	// v3CacheModuleLockRelDirPath is the relative path to the lock files directory for module data.
+	// This directory is used to store lock files for synchronizing reading and writing module data from the cache.
+	//
+	// Normalized.
+	v3CacheModuleLockRelDirPath = normalpath.Join("v3", "modulelocks")
 )
 
 // NewModuleDataProvider returns a new ModuleDataProvider while creating the
@@ -166,12 +173,20 @@ func newModuleDataProvider(
 	if err != nil {
 		return nil, err
 	}
+	if err := createCacheDir(container.CacheDirPath(), v3CacheModuleLockRelDirPath); err != nil {
+		return nil, err
+	}
+	filelocker, err := filelock.NewLocker(normalpath.Join(container.CacheDirPath(), 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;`,
 				),
 			},