repository: Fix indetermenistic voting order in ReplicateRepository

When replicating repositories we not only synchronize the repo's
references, but also have to synchronize the gitconfig and gitattributes
files. To speed up replication a bit we perform these tasks in parallel.
This has the consequence that because all of the tasks perform voting on
the data they're about to change, the order in which the votes arrive at
Praefect is not deterministic. As a result, we see that transactions get
cancelled.

Fix this bug by not running those tasks in parallel anymore. This is a
premature optimization anyway: neither the gitconfig nor gitattributes
files should be big in the general case, so replicating those files
should be a matter of milliseconds. And this is supported by production
data: P99 of the GetConfig RPC is 1.13ms, and GetInfoAttributes has a
P99 of 0.98ms. In contrast to that, replicating the references and
objects is ranging in the hundreds of milliseconds or even dozens of
seconds. Saving about two milliseconds there just isn't worth it.

Changelog: fixed

2 files changed, 42 insertions, 23 deletions

diff --git a/internal/gitaly/service/repository/replicate.go b/internal/gitaly/service/repository/replicate.go
index 13b96cf37..44302e301 100644
--- a/internal/gitaly/service/repository/replicate.go
+++ b/internal/gitaly/service/repository/replicate.go
@@ -25,7 +25,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/internal/tempdir"
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/v14/streamio"
-	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -74,28 +73,18 @@ func (s *server) ReplicateRepository(ctx context.Context, in *gitalypb.Replicate
 		return nil, helper.ErrNotFoundf("source repository does not exist")
 	}
 
-	// We're not using the context of the errgroup here, as an error
-	// returned by either of the called functions would cancel the
-	// respective other function. Given that we're doing RPC calls in
-	// them, cancellation of the calls would mean that the remote side
-	// may still modify the repository even though the local side has
-	// returned already.
-	g, _ := errgroup.WithContext(ctx)
 	outgoingCtx := metadata.IncomingToOutgoing(ctx)
 
-	syncFuncs := []func(context.Context, *gitalypb.ReplicateRepositoryRequest) error{
-		s.syncGitconfig,
-		s.syncInfoAttributes,
-		s.syncRepository,
+	if err := s.syncGitconfig(outgoingCtx, in); err != nil {
+		return nil, helper.ErrInternalf("synchronizing gitconfig: %w", err)
 	}
 
-	for _, f := range syncFuncs {
-		f := f // rescoping f
-		g.Go(func() error { return f(outgoingCtx, in) })
+	if err := s.syncInfoAttributes(outgoingCtx, in); err != nil {
+		return nil, helper.ErrInternalf("synchronizing gitattributes: %w", err)
 	}
 
-	if err := g.Wait(); err != nil {
-		return nil, helper.ErrInternal(err)
+	if err := s.syncRepository(outgoingCtx, in); err != nil {
+		return nil, helper.ErrInternalf("synchronizing repository: %w", err)
 	}
 
 	return &gitalypb.ReplicateRepositoryResponse{}, nil
diff --git a/internal/gitaly/service/repository/replicate_test.go b/internal/gitaly/service/repository/replicate_test.go
index a9b4eba44..a99aa1ac0 100644
--- a/internal/gitaly/service/repository/replicate_test.go
+++ b/internal/gitaly/service/repository/replicate_test.go
@@ -3,12 +3,13 @@ package repository
 import (
 	"bytes"
 	"context"
+	"crypto/sha1"
+	"encoding/hex"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
-	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -133,10 +134,10 @@ func testReplicateRepositoryTransactional(t *testing.T, ctx context.Context) {
 	targetRepo := proto.Clone(sourceRepo).(*gitalypb.Repository)
 	targetRepo.StorageName = cfg.Storages[1].Name
 
-	votes := int32(0)
+	var votes []string
 	txServer := testTransactionServer{
 		vote: func(request *gitalypb.VoteTransactionRequest) (*gitalypb.VoteTransactionResponse, error) {
-			atomic.AddInt32(&votes, 1)
+			votes = append(votes, hex.EncodeToString(request.ReferenceUpdatesHash))
 			return &gitalypb.VoteTransactionResponse{
 				State: gitalypb.VoteTransactionResponse_COMMIT,
 			}, nil
@@ -165,13 +166,35 @@ func testReplicateRepositoryTransactional(t *testing.T, ctx context.Context) {
 	})
 	require.NoError(t, err)
 
-	require.EqualValues(t, 6, atomic.LoadInt32(&votes))
+	// There is no gitattributes file, so we vote on the empty contents of that file.
+	gitattributesVote := sha1.Sum([]byte{})
+	// There is a gitconfig though, so the vote should reflect its contents.
+	gitconfigVote := sha1.Sum(testhelper.MustReadFile(t, filepath.Join(sourceRepoPath, "config")))
+
+	require.Equal(t, []string{
+		// We cannot easily derive these first two votes: they are based on the complete
+		// hashed contents of the unpacked repository. We thus just only assert that they
+		// are always the first two entries and that they are the same by simply taking the
+		// first vote twice here.
+		votes[0],
+		votes[0],
+		hex.EncodeToString(gitconfigVote[:]),
+		hex.EncodeToString(gitconfigVote[:]),
+		hex.EncodeToString(gitattributesVote[:]),
+		hex.EncodeToString(gitattributesVote[:]),
+	}, votes)
+
+	// We're about to change refs/heads/master, and thus the mirror-fetch will update it. The
+	// vote should reflect that.
+	oldOID := text.ChompBytes(gittest.Exec(t, cfg, "-C", sourceRepoPath, "rev-parse", "refs/heads/master"))
+	newOID := text.ChompBytes(gittest.Exec(t, cfg, "-C", sourceRepoPath, "rev-parse", "refs/heads/master~"))
+	replicationVote := sha1.Sum([]byte(fmt.Sprintf("%[1]s %[2]s refs/heads/master\n%[1]s %[2]s HEAD\n", oldOID, newOID)))
 
 	// We're now changing a reference in the source repository such that we can observe changes
 	// in the target repo.
 	gittest.Exec(t, cfg, "-C", sourceRepoPath, "update-ref", "refs/heads/master", "refs/heads/master~")
 
-	atomic.StoreInt32(&votes, 0)
+	votes = nil
 
 	// And the second invocation uses FetchInternalRemote.
 	_, err = client.ReplicateRepository(ctx, &gitalypb.ReplicateRepositoryRequest{
@@ -179,7 +202,14 @@ func testReplicateRepositoryTransactional(t *testing.T, ctx context.Context) {
 		Source:     sourceRepo,
 	})
 	require.NoError(t, err)
-	require.EqualValues(t, 6, atomic.LoadInt32(&votes))
+	require.Equal(t, []string{
+		hex.EncodeToString(gitconfigVote[:]),
+		hex.EncodeToString(gitconfigVote[:]),
+		hex.EncodeToString(gitattributesVote[:]),
+		hex.EncodeToString(gitattributesVote[:]),
+		hex.EncodeToString(replicationVote[:]),
+		hex.EncodeToString(replicationVote[:]),
+	}, votes)
 }
 
 func TestReplicateRepositoryInvalidArguments(t *testing.T) {