Merge branch 'pks-transaction-operations' into 'master'

Transaction support for branches and tags via operations service

See merge request gitlab-org/gitaly!2374

9 files changed, 219 insertions, 168 deletions

diff --git a/changelogs/unreleased/pks-transaction-operations.yml b/changelogs/unreleased/pks-transaction-operations.yml
new file mode 100644
index 000000000..778754307
--- /dev/null
+++ b/changelogs/unreleased/pks-transaction-operations.yml
@@ -0,0 +1,5 @@
+---
+title: Transaction support for modification of branches and tags via operations service
+merge_request: 2374
+author:
+type: added
diff --git a/internal/praefect/coordinator.go b/internal/praefect/coordinator.go
index 274ea6419..875f26f21 100644
--- a/internal/praefect/coordinator.go
+++ b/internal/praefect/coordinator.go
@@ -159,6 +159,10 @@ func (c *Coordinator) accessorStreamParameters(ctx context.Context, call grpcCal
 
 var transactionRPCs = map[string]featureflag.FeatureFlag{
 	"/gitaly.OperationService/UserCreateBranch": featureflag.ReferenceTransactionsOperationService,
+	"/gitaly.OperationService/UserCreateTag":    featureflag.ReferenceTransactionsOperationService,
+	"/gitaly.OperationService/UserDeleteBranch": featureflag.ReferenceTransactionsOperationService,
+	"/gitaly.OperationService/UserDeleteTag":    featureflag.ReferenceTransactionsOperationService,
+	"/gitaly.OperationService/UserUpdateBranch": featureflag.ReferenceTransactionsOperationService,
 	"/gitaly.SSHService/SSHReceivePack":         featureflag.ReferenceTransactionsSSHService,
 	"/gitaly.SmartHTTPService/PostReceivePack":  featureflag.ReferenceTransactionsSmartHTTPService,
 }
diff --git a/internal/service/operations/branches_test.go b/internal/service/operations/branches_test.go
index cbc90472d..8720693dd 100644
--- a/internal/service/operations/branches_test.go
+++ b/internal/service/operations/branches_test.go
@@ -225,35 +225,44 @@ func TestFailedUserCreateBranchRequest(t *testing.T) {
 }
 
 func TestSuccessfulUserDeleteBranchRequest(t *testing.T) {
-	ctx, cancel := testhelper.Context()
-	defer cancel()
+	featureSets, err := testhelper.NewFeatureSets([]featureflag.FeatureFlag{featureflag.ReferenceTransactions})
+	require.NoError(t, err)
 
-	testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
-	defer cleanupFn()
+	for _, featureSet := range featureSets {
+		t.Run(featureSet.String(), func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
 
-	serverSocketPath, stop := runOperationServiceServer(t)
-	defer stop()
+			ctx = featureSet.WithParent(ctx)
 
-	client, conn := newOperationClient(t, serverSocketPath)
-	defer conn.Close()
+			testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
+			defer cleanupFn()
 
-	branchNameInput := "to-be-deleted-soon-branch"
+			serverSocketPath, stop := runOperationServiceServer(t)
+			defer stop()
 
-	defer exec.Command("git", "-C", testRepoPath, "branch", "-d", branchNameInput).Run()
+			client, conn := newOperationClient(t, serverSocketPath)
+			defer conn.Close()
 
-	testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "branch", branchNameInput)
+			branchNameInput := "to-be-deleted-soon-branch"
 
-	request := &gitalypb.UserDeleteBranchRequest{
-		Repository: testRepo,
-		BranchName: []byte(branchNameInput),
-		User:       testhelper.TestUser,
-	}
+			defer exec.Command("git", "-C", testRepoPath, "branch", "-d", branchNameInput).Run()
 
-	_, err := client.UserDeleteBranch(ctx, request)
-	require.NoError(t, err)
+			testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "branch", branchNameInput)
+
+			request := &gitalypb.UserDeleteBranchRequest{
+				Repository: testRepo,
+				BranchName: []byte(branchNameInput),
+				User:       testhelper.TestUser,
+			}
+
+			_, err := client.UserDeleteBranch(ctx, request)
+			require.NoError(t, err)
 
-	branches := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "branch")
-	require.NotContains(t, string(branches), branchNameInput, "branch name still exists in branches list")
+			branches := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "branch")
+			require.NotContains(t, string(branches), branchNameInput, "branch name still exists in branches list")
+		})
+	}
 }
 
 func TestSuccessfulGitHooksForUserDeleteBranchRequest(t *testing.T) {
diff --git a/internal/service/operations/tags_test.go b/internal/service/operations/tags_test.go
index aeaa73d6a..620ba537b 100644
--- a/internal/service/operations/tags_test.go
+++ b/internal/service/operations/tags_test.go
@@ -11,41 +11,51 @@ import (
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/git/log"
 	"gitlab.com/gitlab-org/gitaly/internal/helper/text"
+	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"google.golang.org/grpc/codes"
 )
 
 func TestSuccessfulUserDeleteTagRequest(t *testing.T) {
-	ctx, cancel := testhelper.Context()
-	defer cancel()
+	featureSets, err := testhelper.NewFeatureSets([]featureflag.FeatureFlag{featureflag.ReferenceTransactions})
+	require.NoError(t, err)
 
-	serverSocketPath, stop := runOperationServiceServer(t)
-	defer stop()
+	for _, featureSet := range featureSets {
+		t.Run(featureSet.String(), func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
 
-	client, conn := newOperationClient(t, serverSocketPath)
-	defer conn.Close()
+			ctx = featureSet.WithParent(ctx)
 
-	testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
-	defer cleanupFn()
+			serverSocketPath, stop := runOperationServiceServer(t)
+			defer stop()
 
-	tagNameInput := "to-be-deleted-soon-tag"
+			client, conn := newOperationClient(t, serverSocketPath)
+			defer conn.Close()
 
-	defer exec.Command("git", "-C", testRepoPath, "tag", "-d", tagNameInput).Run()
+			testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
+			defer cleanupFn()
 
-	testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag", tagNameInput)
+			tagNameInput := "to-be-deleted-soon-tag"
 
-	request := &gitalypb.UserDeleteTagRequest{
-		Repository: testRepo,
-		TagName:    []byte(tagNameInput),
-		User:       testhelper.TestUser,
-	}
+			defer exec.Command("git", "-C", testRepoPath, "tag", "-d", tagNameInput).Run()
 
-	_, err := client.UserDeleteTag(ctx, request)
-	require.NoError(t, err)
+			testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag", tagNameInput)
 
-	tags := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag")
-	require.NotContains(t, string(tags), tagNameInput, "tag name still exists in tags list")
+			request := &gitalypb.UserDeleteTagRequest{
+				Repository: testRepo,
+				TagName:    []byte(tagNameInput),
+				User:       testhelper.TestUser,
+			}
+
+			_, err := client.UserDeleteTag(ctx, request)
+			require.NoError(t, err)
+
+			tags := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag")
+			require.NotContains(t, string(tags), tagNameInput, "tag name still exists in tags list")
+		})
+	}
 }
 
 func TestSuccessfulGitHooksForUserDeleteTagRequest(t *testing.T) {
@@ -91,97 +101,106 @@ func testSuccessfulGitHooksForUserDeleteTagRequest(t *testing.T, ctx context.Con
 }
 
 func TestSuccessfulUserCreateTagRequest(t *testing.T) {
-	ctx, cancel := testhelper.Context()
-	defer cancel()
+	featureSets, err := testhelper.NewFeatureSets([]featureflag.FeatureFlag{featureflag.ReferenceTransactions})
+	require.NoError(t, err)
 
-	serverSocketPath, stop := runOperationServiceServer(t)
-	defer stop()
+	for _, featureSet := range featureSets {
+		t.Run(featureSet.String(), func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
 
-	client, conn := newOperationClient(t, serverSocketPath)
-	defer conn.Close()
+			ctx = featureSet.WithParent(ctx)
 
-	testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
-	defer cleanupFn()
+			serverSocketPath, stop := runOperationServiceServer(t)
+			defer stop()
 
-	targetRevision := "c7fbe50c7c7419d9701eebe64b1fdacc3df5b9dd"
-	targetRevisionCommit, err := log.GetCommit(ctx, testRepo, targetRevision)
-	require.NoError(t, err)
+			client, conn := newOperationClient(t, serverSocketPath)
+			defer conn.Close()
 
-	inputTagName := "to-be-créated-soon"
+			testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
+			defer cleanupFn()
 
-	cwd, err := os.Getwd()
-	require.NoError(t, err)
-	preReceiveHook := filepath.Join(cwd, "testdata/pre-receive-expect-object-type")
-	updateHook := filepath.Join(cwd, "testdata/update-expect-object-type")
+			targetRevision := "c7fbe50c7c7419d9701eebe64b1fdacc3df5b9dd"
+			targetRevisionCommit, err := log.GetCommit(ctx, testRepo, targetRevision)
+			require.NoError(t, err)
 
-	testCases := []struct {
-		desc               string
-		tagName            string
-		message            string
-		targetRevision     string
-		expectedTag        *gitalypb.Tag
-		expectedObjectType string
-	}{
-		{
-			desc:           "lightweight tag",
-			tagName:        inputTagName,
-			targetRevision: targetRevision,
-			expectedTag: &gitalypb.Tag{
-				Name:         []byte(inputTagName),
-				TargetCommit: targetRevisionCommit,
-			},
-			expectedObjectType: "commit",
-		},
-		{
-			desc:           "annotated tag",
-			tagName:        inputTagName,
-			targetRevision: targetRevision,
-			message:        "This is an annotated tag",
-			expectedTag: &gitalypb.Tag{
-				Name:         []byte(inputTagName),
-				TargetCommit: targetRevisionCommit,
-				Message:      []byte("This is an annotated tag"),
-				MessageSize:  24,
-			},
-			expectedObjectType: "tag",
-		},
-	}
+			inputTagName := "to-be-créated-soon"
 
-	for _, testCase := range testCases {
-		t.Run(testCase.desc, func(t *testing.T) {
-			for hook, content := range map[string]string{
-				"pre-receive": fmt.Sprintf("#!/bin/sh\n%s %s \"$@\"", preReceiveHook, testCase.expectedObjectType),
-				"update":      fmt.Sprintf("#!/bin/sh\n%s %s \"$@\"", updateHook, testCase.expectedObjectType),
-			} {
-				hookCleanup, err := testhelper.WriteCustomHook(testRepoPath, hook, []byte(content))
-				require.NoError(t, err)
-				defer hookCleanup()
+			cwd, err := os.Getwd()
+			require.NoError(t, err)
+			preReceiveHook := filepath.Join(cwd, "testdata/pre-receive-expect-object-type")
+			updateHook := filepath.Join(cwd, "testdata/update-expect-object-type")
+
+			testCases := []struct {
+				desc               string
+				tagName            string
+				message            string
+				targetRevision     string
+				expectedTag        *gitalypb.Tag
+				expectedObjectType string
+			}{
+				{
+					desc:           "lightweight tag",
+					tagName:        inputTagName,
+					targetRevision: targetRevision,
+					expectedTag: &gitalypb.Tag{
+						Name:         []byte(inputTagName),
+						TargetCommit: targetRevisionCommit,
+					},
+					expectedObjectType: "commit",
+				},
+				{
+					desc:           "annotated tag",
+					tagName:        inputTagName,
+					targetRevision: targetRevision,
+					message:        "This is an annotated tag",
+					expectedTag: &gitalypb.Tag{
+						Name:         []byte(inputTagName),
+						TargetCommit: targetRevisionCommit,
+						Message:      []byte("This is an annotated tag"),
+						MessageSize:  24,
+					},
+					expectedObjectType: "tag",
+				},
 			}
 
-			request := &gitalypb.UserCreateTagRequest{
-				Repository:     testRepo,
-				TagName:        []byte(testCase.tagName),
-				TargetRevision: []byte(testCase.targetRevision),
-				User:           testhelper.TestUser,
-				Message:        []byte(testCase.message),
+			for _, testCase := range testCases {
+				t.Run(testCase.desc, func(t *testing.T) {
+					for hook, content := range map[string]string{
+						"pre-receive": fmt.Sprintf("#!/bin/sh\n%s %s \"$@\"", preReceiveHook, testCase.expectedObjectType),
+						"update":      fmt.Sprintf("#!/bin/sh\n%s %s \"$@\"", updateHook, testCase.expectedObjectType),
+					} {
+						hookCleanup, err := testhelper.WriteCustomHook(testRepoPath, hook, []byte(content))
+						require.NoError(t, err)
+						defer hookCleanup()
+					}
+
+					request := &gitalypb.UserCreateTagRequest{
+						Repository:     testRepo,
+						TagName:        []byte(testCase.tagName),
+						TargetRevision: []byte(testCase.targetRevision),
+						User:           testhelper.TestUser,
+						Message:        []byte(testCase.message),
+					}
+
+					ctx, cancel := testhelper.Context()
+					defer cancel()
+
+					response, err := client.UserCreateTag(ctx, request)
+					require.NoError(t, err, "error from calling RPC")
+					require.Empty(t, response.PreReceiveError, "PreReceiveError must be empty, signalling the push was accepted")
+
+					defer exec.Command("git", "-C", testRepoPath, "tag", "-d", inputTagName).Run()
+
+					id := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "rev-parse", inputTagName)
+					testCase.expectedTag.Id = text.ChompBytes(id)
+
+					require.Equal(t, testCase.expectedTag, response.Tag)
+
+					tag := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag")
+					require.Contains(t, string(tag), inputTagName)
+				})
 			}
-
-			ctx, cancel := testhelper.Context()
-			defer cancel()
-
-			response, err := client.UserCreateTag(ctx, request)
-			require.NoError(t, err, "error from calling RPC")
-			require.Empty(t, response.PreReceiveError, "PreReceiveError must be empty, signalling the push was accepted")
-
-			defer exec.Command("git", "-C", testRepoPath, "tag", "-d", inputTagName).Run()
-
-			id := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "rev-parse", inputTagName)
-			testCase.expectedTag.Id = text.ChompBytes(id)
-
-			require.Equal(t, testCase.expectedTag, response.Tag)
-
-			tag := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "tag")
-			require.Contains(t, string(tag), inputTagName)
 		})
 	}
 }
diff --git a/internal/service/operations/update_branches_test.go b/internal/service/operations/update_branches_test.go
index f151d9f69..542d6cff2 100644
--- a/internal/service/operations/update_branches_test.go
+++ b/internal/service/operations/update_branches_test.go
@@ -21,35 +21,44 @@ var (
 )
 
 func TestSuccessfulUserUpdateBranchRequest(t *testing.T) {
-	ctx, cancel := testhelper.Context()
-	defer cancel()
+	featureSets, err := testhelper.NewFeatureSets([]featureflag.FeatureFlag{featureflag.ReferenceTransactions})
+	require.NoError(t, err)
 
-	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
-	defer cleanupFn()
+	for _, featureSet := range featureSets {
+		t.Run(featureSet.String(), func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
 
-	serverSocketPath, stop := runOperationServiceServer(t)
-	defer stop()
+			ctx = featureSet.WithParent(ctx)
 
-	client, conn := newOperationClient(t, serverSocketPath)
-	defer conn.Close()
+			testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+			defer cleanupFn()
 
-	request := &gitalypb.UserUpdateBranchRequest{
-		Repository: testRepo,
-		BranchName: []byte(updateBranchName),
-		Newrev:     newrev,
-		Oldrev:     oldrev,
-		User:       testhelper.TestUser,
-	}
+			serverSocketPath, stop := runOperationServiceServer(t)
+			defer stop()
 
-	response, err := client.UserUpdateBranch(ctx, request)
+			client, conn := newOperationClient(t, serverSocketPath)
+			defer conn.Close()
 
-	require.NoError(t, err)
-	require.Empty(t, response.PreReceiveError)
+			request := &gitalypb.UserUpdateBranchRequest{
+				Repository: testRepo,
+				BranchName: []byte(updateBranchName),
+				Newrev:     newrev,
+				Oldrev:     oldrev,
+				User:       testhelper.TestUser,
+			}
 
-	branchCommit, err := log.GetCommit(ctx, testRepo, updateBranchName)
+			response, err := client.UserUpdateBranch(ctx, request)
 
-	require.NoError(t, err)
-	require.Equal(t, string(newrev), branchCommit.Id)
+			require.NoError(t, err)
+			require.Empty(t, response.PreReceiveError)
+
+			branchCommit, err := log.GetCommit(ctx, testRepo, updateBranchName)
+
+			require.NoError(t, err)
+			require.Equal(t, string(newrev), branchCommit.Id)
+		})
+	}
 }
 
 func TestSuccessfulGitHooksForUserUpdateBranchRequest(t *testing.T) {
diff --git a/ruby/lib/gitaly_server/operations_service.rb b/ruby/lib/gitaly_server/operations_service.rb
index 0f4db6c7d..a6cd3f13d 100644
--- a/ruby/lib/gitaly_server/operations_service.rb
+++ b/ruby/lib/gitaly_server/operations_service.rb
@@ -4,6 +4,7 @@ module GitalyServer
 
     def user_create_tag(request, call)
       repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
+      transaction = Praefect::Transaction.from_metadata(call.metadata)
 
       gitaly_user = get_param!(request, :user)
       user = Gitlab::Git::User.from_gitaly(gitaly_user)
@@ -12,7 +13,7 @@ module GitalyServer
 
       target_revision = get_param!(request, :target_revision)
 
-      created_tag = repo.add_tag(tag_name, user: user, target: target_revision, message: request.message.presence)
+      created_tag = repo.add_tag(tag_name, user: user, target: target_revision, message: request.message.presence, transaction: transaction)
       Gitaly::UserCreateTagResponse.new unless created_tag
 
       rugged_commit = created_tag.dereferenced_target.rugged_commit
@@ -30,13 +31,14 @@ module GitalyServer
 
     def user_delete_tag(request, call)
       repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
+      transaction = Praefect::Transaction.from_metadata(call.metadata)
 
       gitaly_user = get_param!(request, :user)
       user = Gitlab::Git::User.from_gitaly(gitaly_user)
 
       tag_name = get_param!(request, :tag_name)
 
-      repo.rm_tag(tag_name, user: user)
+      repo.rm_tag(tag_name, user: user, transaction: transaction)
 
       Gitaly::UserDeleteTagResponse.new
     rescue Gitlab::Git::PreReceiveError => e
@@ -72,9 +74,10 @@ module GitalyServer
       newrev = get_param!(request, :newrev)
       oldrev = get_param!(request, :oldrev)
       gitaly_user = get_param!(request, :user)
+      transaction = Praefect::Transaction.from_metadata(call.metadata)
 
       user = Gitlab::Git::User.from_gitaly(gitaly_user)
-      repo.update_branch(branch_name, user: user, newrev: newrev, oldrev: oldrev)
+      repo.update_branch(branch_name, user: user, newrev: newrev, oldrev: oldrev, transaction: transaction)
 
       Gitaly::UserUpdateBranchResponse.new
     rescue Gitlab::Git::Repository::InvalidRef, Gitlab::Git::CommitError => ex
@@ -86,8 +89,9 @@ module GitalyServer
     def user_delete_branch(request, call)
       repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
       user = Gitlab::Git::User.from_gitaly(request.user)
+      transaction = Praefect::Transaction.from_metadata(call.metadata)
 
-      repo.rm_branch(request.branch_name, user: user)
+      repo.rm_branch(request.branch_name, user: user, transaction: transaction)
 
       Gitaly::UserDeleteBranchResponse.new
     rescue Gitlab::Git::PreReceiveError => e
diff --git a/ruby/lib/gitlab/git/operation_service.rb b/ruby/lib/gitlab/git/operation_service.rb
index e6025b673..e6379493b 100644
--- a/ruby/lib/gitlab/git/operation_service.rb
+++ b/ruby/lib/gitlab/git/operation_service.rb
@@ -32,35 +32,35 @@ module Gitlab
         update_ref_in_hooks(ref, newrev, oldrev, transaction: transaction)
       end
 
-      def rm_branch(branch)
+      def rm_branch(branch, transaction: nil)
         ref = Gitlab::Git::BRANCH_REF_PREFIX + branch.name
         oldrev = branch.target
         newrev = Gitlab::Git::BLANK_SHA
 
-        update_ref_in_hooks(ref, newrev, oldrev)
+        update_ref_in_hooks(ref, newrev, oldrev, transaction: transaction)
       end
 
-      def add_lightweight_tag(tag_name, tag_target)
+      def add_lightweight_tag(tag_name, tag_target, transaction: nil)
         ref = Gitlab::Git::TAG_REF_PREFIX + tag_name
         oldrev = Gitlab::Git::BLANK_SHA
 
-        update_ref_in_hooks(ref, tag_target, oldrev)
+        update_ref_in_hooks(ref, tag_target, oldrev, transaction: transaction)
       end
 
-      def add_annotated_tag(tag_name, tag_target, options)
+      def add_annotated_tag(tag_name, tag_target, options, transaction: nil)
         ref = Gitlab::Git::TAG_REF_PREFIX + tag_name
         oldrev = Gitlab::Git::BLANK_SHA
         annotation = repository.rugged.tags.create_annotation(tag_name, tag_target, options)
 
-        update_ref_in_hooks(ref, annotation.oid, oldrev)
+        update_ref_in_hooks(ref, annotation.oid, oldrev, transaction: transaction)
       end
 
-      def rm_tag(tag)
+      def rm_tag(tag, transaction: nil)
         ref = Gitlab::Git::TAG_REF_PREFIX + tag.name
         oldrev = tag.target
         newrev = Gitlab::Git::BLANK_SHA
 
-        update_ref_in_hooks(ref, newrev, oldrev) do
+        update_ref_in_hooks(ref, newrev, oldrev, transaction: transaction) do
           repository.rugged.tags.delete(tag_name)
         end
       end
@@ -96,9 +96,9 @@ module Gitlab
         end
       end
 
-      def update_branch(branch_name, newrev, oldrev, push_options: nil)
+      def update_branch(branch_name, newrev, oldrev, push_options: nil, transaction: nil)
         ref = Gitlab::Git::BRANCH_REF_PREFIX + branch_name
-        update_ref_in_hooks(ref, newrev, oldrev, push_options: push_options)
+        update_ref_in_hooks(ref, newrev, oldrev, push_options: push_options, transaction: transaction)
       end
 
       # Yields the given block (which should return a commit) and
diff --git a/ruby/lib/gitlab/git/repository.rb b/ruby/lib/gitlab/git/repository.rb
index 675f4b2e2..540e61a18 100644
--- a/ruby/lib/gitlab/git/repository.rb
+++ b/ruby/lib/gitlab/git/repository.rb
@@ -222,7 +222,7 @@ module Gitlab
         false
       end
 
-      def add_tag(tag_name, user:, target:, message: nil)
+      def add_tag(tag_name, user:, target:, message: nil, transaction: nil)
         target_object = Ref.dereference_object(lookup(target))
         raise InvalidRef, "target not found: #{target}" unless target_object
 
@@ -234,10 +234,11 @@ module Gitlab
             tag_name,
             target_oid,
             message: message,
-            tagger: Gitlab::Git.committer_hash(email: user.email, name: user.name)
+            tagger: Gitlab::Git.committer_hash(email: user.email, name: user.name),
+            transaction: transaction
           )
         else
-          operation_service.add_lightweight_tag(tag_name, target_oid)
+          operation_service.add_lightweight_tag(tag_name, target_oid, transaction: transaction)
         end
 
         find_tag(tag_name)
@@ -251,24 +252,24 @@ module Gitlab
         end
       end
 
-      def update_branch(branch_name, user:, newrev:, oldrev:, push_options: nil)
-        OperationService.new(user, self).update_branch(branch_name, newrev, oldrev, push_options: push_options)
+      def update_branch(branch_name, user:, newrev:, oldrev:, push_options: nil, transaction: nil)
+        OperationService.new(user, self).update_branch(branch_name, newrev, oldrev, push_options: push_options, transaction: transaction)
       end
 
-      def rm_branch(branch_name, user:)
+      def rm_branch(branch_name, user:, transaction: nil)
         branch = find_branch(branch_name)
 
         raise InvalidRef, "branch not found: #{branch_name}" unless branch
 
-        OperationService.new(user, self).rm_branch(branch)
+        OperationService.new(user, self).rm_branch(branch, transaction: transaction)
       end
 
-      def rm_tag(tag_name, user:)
+      def rm_tag(tag_name, user:, transaction: nil)
         tag = find_tag(tag_name)
 
         raise InvalidRef, "tag not found: #{tag_name}" unless tag
 
-        Gitlab::Git::OperationService.new(user, self).rm_tag(tag)
+        Gitlab::Git::OperationService.new(user, self).rm_tag(tag, transaction: transaction)
       end
 
       def find_tag(name)
diff --git a/ruby/lib/praefect/transaction.rb b/ruby/lib/praefect/transaction.rb
index 3ddf6c5f6..608839a48 100644
--- a/ruby/lib/praefect/transaction.rb
+++ b/ruby/lib/praefect/transaction.rb
@@ -1,9 +1,9 @@
 module Praefect
   class Transaction
-    PRAEFECT_SERVER_KEY = "praefect-server".freeze
-    PRAEFECT_SERVER_ENV = "PRAEFECT_SERVER".freeze
-    TRANSACTION_KEY = "transaction".freeze
-    TRANSACTION_ENV = "REFERENCE_TRANSACTION".freeze
+    PRAEFECT_SERVER_KEY = "gitaly-praefect-server".freeze
+    PRAEFECT_SERVER_ENV = "GITALY_PRAEFECT_SERVER".freeze
+    TRANSACTION_KEY = "gitaly-reference-transaction".freeze
+    TRANSACTION_ENV = "GITALY_REFERENCE_TRANSACTION".freeze
 
     def self.from_metadata(metadata)
       new(metadata[PRAEFECT_SERVER_KEY], metadata[TRANSACTION_KEY])