From 8bc9f82a3e2e1f9ab3b47172c8068c1bdbe19bbc Mon Sep 17 00:00:00 2001
From: John Cai <jcai@gitlab.com>
Date: Fri, 7 Feb 2020 12:32:46 -0800
Subject: [PATCH] Lock RepolicateRepository to 1 operation at a time

---
 internal/service/repository/replicate.go      | 19 +++++
 internal/service/repository/replicate_test.go | 72 +++++++++++++++++++
 .../service/repository/testhelper_test.go     |  4 ++
 3 files changed, 95 insertions(+)

diff --git a/internal/service/repository/replicate.go b/internal/service/repository/replicate.go
index 9d0d2330007..ef09b979218 100644
--- a/internal/service/repository/replicate.go
+++ b/internal/service/repository/replicate.go
@@ -23,6 +23,10 @@ import (
 	"google.golang.org/grpc"
 )
 
+const (
+	replicationLockFileName = "replication.lock"
+)
+
 func (s *server) ReplicateRepository(ctx context.Context, in *gitalypb.ReplicateRepositoryRequest) (*gitalypb.ReplicateRepositoryResponse, error) {
 	if err := validateReplicateRepository(in); err != nil {
 		return nil, helper.ErrInvalidArgument(err)
@@ -37,6 +41,21 @@ func (s *server) ReplicateRepository(ctx context.Context, in *gitalypb.Replicate
 		return nil, helper.ErrInternal(err)
 	}
 
+	// hold lockfile
+	lockfilePath := filepath.Join(repoPath, replicationLockFileName)
+	lockFile, err := os.OpenFile(lockfilePath, os.O_EXCL|os.O_CREATE, 0)
+	if err != nil {
+		if os.IsExist(err) {
+			return &gitalypb.ReplicateRepositoryResponse{}, nil
+		}
+		return nil, helper.ErrInternal(err)
+	}
+
+	defer func() {
+		os.Remove(lockfilePath)
+		lockFile.Close()
+	}()
+
 	if helper.IsGitDirectory(repoPath) {
 		syncFuncs = append(syncFuncs, s.syncRepository)
 	} else {
diff --git a/internal/service/repository/replicate_test.go b/internal/service/repository/replicate_test.go
index 2824f3f6873..2ae48c3053e 100644
--- a/internal/service/repository/replicate_test.go
+++ b/internal/service/repository/replicate_test.go
@@ -248,3 +248,75 @@ func TestReplicateRepository_BadRepository(t *testing.T) {
 
 	testhelper.MustRunCommand(t, nil, "git", "-C", targetRepoPath, "fsck")
 }
+
+func TestReplicateRepositoryConcurrent(t *testing.T) {
+	tmpPath, cleanup := testhelper.TempDir(t, t.Name())
+	defer cleanup()
+
+	replicaPath := filepath.Join(tmpPath, "replica")
+	require.NoError(t, os.MkdirAll(replicaPath, 0755))
+
+	defer func(storages []config.Storage) {
+		config.Config.Storages = storages
+	}(config.Config.Storages)
+
+	config.Config.Storages = []config.Storage{
+		config.Storage{
+			Name: "default",
+			Path: testhelper.GitlabTestStoragePath(),
+		},
+		config.Storage{
+			Name: "replica",
+			Path: replicaPath,
+		},
+	}
+
+	server, serverSocketPath := runFullServer(t)
+	defer server.Stop()
+
+	testRepo, _, cleanupRepo := testhelper.NewTestRepo(t)
+	defer cleanupRepo()
+
+	config.Config.SocketPath = serverSocketPath
+
+	repoClient, conn := repository.NewRepositoryClient(t, serverSocketPath)
+	defer conn.Close()
+
+	targetRepo := *testRepo
+	targetRepo.StorageName = "replica"
+
+	targetRepoPath, err := helper.GetPath(&targetRepo)
+	require.NoError(t, err)
+
+	require.NoError(t, os.MkdirAll(targetRepoPath, 0755))
+
+	lockfilePath := filepath.Join(targetRepoPath, repository.ReplicationLockFileName)
+
+	// write a lockfile in the repository directory
+	f, err := os.OpenFile(lockfilePath, os.O_EXCL|os.O_CREATE, 0)
+	require.NoError(t, err)
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+	md := testhelper.GitalyServersMetadata(t, serverSocketPath)
+	injectedCtx := metadata.NewOutgoingContext(ctx, md)
+
+	_, err = repoClient.ReplicateRepository(injectedCtx, &gitalypb.ReplicateRepositoryRequest{
+		Repository: &targetRepo,
+		Source:     testRepo,
+	})
+	require.NoError(t, err)
+
+	require.False(t, helper.IsGitDirectory(targetRepoPath), "replication should not have happened")
+
+	// remove lockfile
+	require.NoError(t, f.Close())
+	require.NoError(t, os.Remove(lockfilePath))
+
+	_, err = repoClient.ReplicateRepository(injectedCtx, &gitalypb.ReplicateRepositoryRequest{
+		Repository: &targetRepo,
+		Source:     testRepo,
+	})
+	require.NoError(t, err)
+	require.True(t, helper.IsGitDirectory(targetRepoPath), "replication should have happened")
+}
diff --git a/internal/service/repository/testhelper_test.go b/internal/service/repository/testhelper_test.go
index 37c5e6d892e..ea5839c39c8 100644
--- a/internal/service/repository/testhelper_test.go
+++ b/internal/service/repository/testhelper_test.go
@@ -153,3 +153,7 @@ func testMain(m *testing.M) int {
 
 	return m.Run()
 }
+
+const (
+	ReplicationLockFileName = replicationLockFileName
+)
-- 
GitLab