pre-receive: Only execute hooks on primary node

When using reference transactions, we started pushing Git data to both
primary and secondaries at the same time. As a result, git-receive-pack
will be executed on all nodes, which wasn't the case previously. This
leads us to also execute the pre-receive hook on all nodes, which is
definitely unexpected and a change in behaviour, as it would previously
only have executed on the primary node.

Now that we have information available about each node's role, this
commit fixes the logic to only execute both Git hooks as well as the
pre-receive reference counters in case the current Gitaly node is the
transaction's primary.

2 files changed, 226 insertions, 63 deletions

diff --git a/internal/service/hook/pre_receive.go b/internal/service/hook/pre_receive.go
index 5c79c7ca8..9adfa0d8c 100644
--- a/internal/service/hook/pre_receive.go
+++ b/internal/service/hook/pre_receive.go
@@ -172,6 +172,20 @@ func (s *server) executeCustomHooks(stream gitalypb.HookService_PreReceiveHookSe
 	return nil
 }
 
+func isPrimary(env []string) (bool, error) {
+	tx, err := metadata.TransactionFromEnv(env)
+	if err != nil {
+		if errors.Is(err, metadata.ErrTransactionNotFound) {
+			// If there is no transaction, then we only ever write
+			// to the primary. Thus, we return true.
+			return true, nil
+		}
+		return false, err
+	}
+
+	return tx.Primary, nil
+}
+
 func (s *server) PreReceiveHook(stream gitalypb.HookService_PreReceiveHookServer) error {
 	firstRequest, err := stream.Recv()
 	if err != nil {
@@ -200,32 +214,40 @@ func (s *server) PreReceiveHook(stream gitalypb.HookService_PreReceiveHookServer
 
 	glID, glRepo, glProtocol := getEnvVar("GL_ID", reqEnvVars), getEnvVar("GL_REPOSITORY", reqEnvVars), getEnvVar("GL_PROTOCOL", reqEnvVars)
 
-	allowed, message, err := s.gitlabAPI.Allowed(repository, glRepo, glID, glProtocol, string(changes))
+	primary, err := isPrimary(reqEnvVars)
 	if err != nil {
-		return preReceiveHookResponse(stream, int32(1), fmt.Sprintf("GitLab: %v", err))
+		return helper.ErrInternalf("could not check role: %w", err)
 	}
 
-	if !allowed {
-		return preReceiveHookResponse(stream, int32(1), message)
-	}
+	// Only the primary should execute hooks and increment reference counters.
+	if primary {
+		allowed, message, err := s.gitlabAPI.Allowed(repository, glRepo, glID, glProtocol, string(changes))
+		if err != nil {
+			return preReceiveHookResponse(stream, int32(1), fmt.Sprintf("GitLab: %v", err))
+		}
 
-	if err := s.executeCustomHooks(stream, changes, repository, reqEnvVars); err != nil {
-		var exitError *exec.ExitError
-		if errors.As(err, &exitError) {
-			return preReceiveHookResponse(stream, int32(exitError.ExitCode()), "")
+		if !allowed {
+			return preReceiveHookResponse(stream, int32(1), message)
 		}
 
-		return helper.ErrInternal(err)
-	}
+		if err := s.executeCustomHooks(stream, changes, repository, reqEnvVars); err != nil {
+			var exitError *exec.ExitError
+			if errors.As(err, &exitError) {
+				return preReceiveHookResponse(stream, int32(exitError.ExitCode()), "")
+			}
 
-	// reference counter
-	ok, err := s.gitlabAPI.PreReceive(glRepo)
-	if err != nil {
-		return helper.ErrInternalf("calling pre_receive endpoint: %v", err)
-	}
+			return helper.ErrInternal(err)
+		}
 
-	if !ok {
-		return preReceiveHookResponse(stream, 1, "")
+		// reference counter
+		ok, err := s.gitlabAPI.PreReceive(glRepo)
+		if err != nil {
+			return helper.ErrInternalf("calling pre_receive endpoint: %v", err)
+		}
+
+		if !ok {
+			return preReceiveHookResponse(stream, 1, "")
+		}
 	}
 
 	hash := sha1.Sum(changes)
@@ -275,31 +297,42 @@ func (s *server) preReceiveHookRuby(firstRequest *gitalypb.PreReceiveHookRequest
 
 		return stdin, nil
 	})
-	stdout := streamio.NewWriter(func(p []byte) error { return stream.Send(&gitalypb.PreReceiveHookResponse{Stdout: p}) })
-	stderr := streamio.NewWriter(func(p []byte) error { return stream.Send(&gitalypb.PreReceiveHookResponse{Stderr: p}) })
 
-	repoPath, err := helper.GetRepoPath(firstRequest.GetRepository())
+	env, err := preReceiveEnv(firstRequest)
 	if err != nil {
 		return helper.ErrInternal(err)
 	}
 
-	c := exec.Command(gitlabShellHook("pre-receive"))
-	c.Dir = repoPath
-
-	env, err := preReceiveEnv(firstRequest)
+	primary, err := isPrimary(env)
 	if err != nil {
-		return helper.ErrInternal(err)
+		return helper.ErrInternalf("could not check role: %w", err)
 	}
 
-	status, err := streamCommandResponse(
-		stream.Context(),
-		stdin,
-		stdout, stderr,
-		c,
-		env,
-	)
-	if err != nil {
-		return helper.ErrInternal(err)
+	var status int32
+
+	// Only the primary should execute hooks.
+	if primary {
+		stdout := streamio.NewWriter(func(p []byte) error { return stream.Send(&gitalypb.PreReceiveHookResponse{Stdout: p}) })
+		stderr := streamio.NewWriter(func(p []byte) error { return stream.Send(&gitalypb.PreReceiveHookResponse{Stderr: p}) })
+
+		repoPath, err := helper.GetRepoPath(firstRequest.GetRepository())
+		if err != nil {
+			return helper.ErrInternal(err)
+		}
+
+		c := exec.Command(gitlabShellHook("pre-receive"))
+		c.Dir = repoPath
+
+		status, err = streamCommandResponse(
+			stream.Context(),
+			stdin,
+			stdout, stderr,
+			c,
+			env,
+		)
+		if err != nil {
+			return helper.ErrInternal(err)
+		}
 	}
 
 	if err := s.voteOnTransaction(stream, referenceUpdatesHasher.Sum(nil), env); err != nil {
diff --git a/internal/service/hook/pre_receive_test.go b/internal/service/hook/pre_receive_test.go
index e98e44227..e0d3b1f48 100644
--- a/internal/service/hook/pre_receive_test.go
+++ b/internal/service/hook/pre_receive_test.go
@@ -15,6 +15,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/internal/config"
 	"gitlab.com/gitlab-org/gitaly/internal/helper/text"
 	"gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect/metadata"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
@@ -39,7 +40,7 @@ func TestPreReceiveInvalidArgument(t *testing.T) {
 	testhelper.RequireGrpcError(t, err, codes.InvalidArgument)
 }
 
-func receivePreReceive(t *testing.T, stream gitalypb.HookService_PreReceiveHookClient, stdin io.Reader) ([]byte, []byte, int32) {
+func sendPreReceiveHookRequest(t *testing.T, stream gitalypb.HookService_PreReceiveHookClient, stdin io.Reader) ([]byte, []byte, int32, error) {
 	go func() {
 		writer := streamio.NewWriter(func(p []byte) error {
 			return stream.Send(&gitalypb.PreReceiveHookRequest{Stdin: p})
@@ -57,7 +58,7 @@ func receivePreReceive(t *testing.T, stream gitalypb.HookService_PreReceiveHookC
 			break
 		}
 		if err != nil {
-			t.Fatalf("error when receiving stream: %v", err)
+			return stdout.Bytes(), stderr.Bytes(), -1, err
 		}
 
 		_, err = stdout.Write(resp.GetStdout())
@@ -69,7 +70,13 @@ func receivePreReceive(t *testing.T, stream gitalypb.HookService_PreReceiveHookC
 		require.NoError(t, err)
 	}
 
-	return stdout.Bytes(), stderr.Bytes(), status
+	return stdout.Bytes(), stderr.Bytes(), status, nil
+}
+
+func receivePreReceive(t *testing.T, stream gitalypb.HookService_PreReceiveHookClient, stdin io.Reader) ([]byte, []byte, int32) {
+	stdout, stderr, status, err := sendPreReceiveHookRequest(t, stream, stdin)
+	require.NoError(t, err)
+	return stdout, stderr, status
 }
 
 func TestPreReceive(t *testing.T) {
@@ -302,6 +309,27 @@ func TestPreReceiveHook_GitlabAPIAccess(t *testing.T) {
 	assert.Equal(t, "", text.ChompBytes(stdout), "hook stdout")
 }
 
+func preReceiveHandler(increased bool) http.HandlerFunc {
+	return func(res http.ResponseWriter, req *http.Request) {
+		res.Header().Set("Content-Type", "application/json")
+		res.WriteHeader(http.StatusOK)
+		res.Write([]byte(fmt.Sprintf("{\"reference_counter_increased\": %v}", increased)))
+	}
+}
+
+func allowedHandler(allowed bool) http.HandlerFunc {
+	return func(res http.ResponseWriter, req *http.Request) {
+		res.Header().Set("Content-Type", "application/json")
+		if allowed {
+			res.WriteHeader(http.StatusOK)
+			res.Write([]byte(`{"status": true}`))
+		} else {
+			res.WriteHeader(http.StatusUnauthorized)
+			res.Write([]byte(`{"message":"not allowed"}`))
+		}
+	}
+}
+
 func TestPreReceive_APIErrors(t *testing.T) {
 	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
 	defer cleanupFn()
@@ -325,19 +353,9 @@ func TestPreReceive_APIErrors(t *testing.T) {
 			expectedStderr:     "GitLab: not allowed",
 		},
 		{
-			desc: "/pre_receive endpoint fails to increase reference coutner",
-			allowedHandler: http.HandlerFunc(
-				func(res http.ResponseWriter, req *http.Request) {
-					res.Header().Set("Content-Type", "application/json")
-					res.WriteHeader(http.StatusOK)
-					res.Write([]byte(`{"status": true}`))
-				}),
-			preReceiveHandler: http.HandlerFunc(
-				func(res http.ResponseWriter, req *http.Request) {
-					res.Header().Set("Content-Type", "application/json")
-					res.WriteHeader(http.StatusOK)
-					res.Write([]byte(`{"reference_counter_increased": false}`))
-				}),
+			desc:               "/pre_receive endpoint fails to increase reference coutner",
+			allowedHandler:     allowedHandler(true),
+			preReceiveHandler:  preReceiveHandler(false),
 			expectedExitStatus: 1,
 		},
 	}
@@ -399,18 +417,8 @@ func TestPreReceiveHook_CustomHookErrors(t *testing.T) {
 	defer cleanupFn()
 
 	mux := http.NewServeMux()
-	mux.Handle("/api/v4/internal/allowed",
-		http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
-			res.Header().Set("Content-Type", "application/json")
-			res.WriteHeader(http.StatusOK)
-			res.Write([]byte(`{"status": true}`))
-		}))
-	mux.Handle("/api/v4/internal/pre_receive", http.HandlerFunc(
-		func(res http.ResponseWriter, req *http.Request) {
-			res.Header().Set("Content-Type", "application/json")
-			res.WriteHeader(http.StatusOK)
-			res.Write([]byte(`{"reference_counter_increased": true}`))
-		}))
+	mux.Handle("/api/v4/internal/allowed", allowedHandler(true))
+	mux.Handle("/api/v4/internal/pre_receive", preReceiveHandler(true))
 	srv := httptest.NewServer(mux)
 	defer srv.Close()
 
@@ -463,3 +471,125 @@ exit %d
 	require.Equal(t, customHookReturnCode, status)
 	assert.Equal(t, customHookReturnMsg, text.ChompBytes(stderr), "hook stderr")
 }
+
+func TestPreReceiveHook_Primary(t *testing.T) {
+	testCases := []struct {
+		desc               string
+		primary            bool
+		allowedHandler     http.HandlerFunc
+		preReceiveHandler  http.HandlerFunc
+		hookExitCode       int32
+		expectedExitStatus int32
+		expectedStderr     string
+	}{
+		{
+			desc:               "primary checks for permissions",
+			primary:            true,
+			allowedHandler:     allowedHandler(false),
+			expectedExitStatus: 1,
+			expectedStderr:     "GitLab: not allowed",
+		},
+		{
+			desc:               "secondary checks for permissions",
+			primary:            false,
+			allowedHandler:     allowedHandler(false),
+			expectedExitStatus: -1,
+		},
+		{
+			desc:               "primary tries to increase reference counter",
+			primary:            true,
+			allowedHandler:     allowedHandler(true),
+			preReceiveHandler:  preReceiveHandler(false),
+			expectedExitStatus: 1,
+			expectedStderr:     "",
+		},
+		{
+			desc:               "secondary does not try to increase reference counter",
+			primary:            false,
+			allowedHandler:     allowedHandler(true),
+			preReceiveHandler:  preReceiveHandler(false),
+			expectedExitStatus: -1,
+		},
+		{
+			desc:               "primary executes hook",
+			primary:            true,
+			allowedHandler:     allowedHandler(true),
+			preReceiveHandler:  preReceiveHandler(true),
+			hookExitCode:       123,
+			expectedExitStatus: 123,
+		},
+		{
+			desc:               "secondary does not execute hook",
+			primary:            false,
+			allowedHandler:     allowedHandler(true),
+			preReceiveHandler:  preReceiveHandler(true),
+			hookExitCode:       123,
+			expectedExitStatus: -1,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepo(t)
+			defer cleanupFn()
+
+			mux := http.NewServeMux()
+			mux.Handle("/api/v4/internal/allowed", tc.allowedHandler)
+			mux.Handle("/api/v4/internal/pre_receive", tc.preReceiveHandler)
+			srv := httptest.NewServer(mux)
+			defer srv.Close()
+
+			tmpDir, cleanup := testhelper.TempDir(t)
+			defer cleanup()
+
+			secretFilePath := filepath.Join(tmpDir, ".gitlab_shell_secret")
+			testhelper.WriteShellSecretFile(t, tmpDir, "token")
+			testhelper.WriteCustomHook(testRepoPath, "pre-receive", []byte(fmt.Sprintf("#!/bin/bash\nexit %d", tc.hookExitCode)))
+
+			gitlabAPI, err := NewGitlabAPI(config.Gitlab{
+				URL:        srv.URL,
+				SecretFile: secretFilePath,
+			})
+			require.NoError(t, err)
+
+			serverSocketPath, stop := runHooksServerWithAPI(t, gitlabAPI, config.Hooks{})
+			defer stop()
+
+			client, conn := newHooksClient(t, serverSocketPath)
+			defer conn.Close()
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			transaction := metadata.Transaction{
+				ID:      1234,
+				Node:    "node-1",
+				Primary: tc.primary,
+			}
+			transactionEnv, err := transaction.Env()
+			require.NoError(t, err)
+
+			environment := []string{
+				"GL_ID=key-123",
+				"GL_PROTOCOL=web",
+				"GL_USERNAME=username",
+				"GL_REPOSITORY=repository",
+				fmt.Sprintf("%s=true", featureflag.GoPreReceiveHookEnvVar),
+				transactionEnv,
+			}
+
+			stream, err := client.PreReceiveHook(ctx)
+			require.NoError(t, err)
+			require.NoError(t, stream.Send(&gitalypb.PreReceiveHookRequest{
+				Repository:           testRepo,
+				EnvironmentVariables: environment,
+			}))
+			require.NoError(t, stream.CloseSend())
+
+			_, stderr, status, _ := sendPreReceiveHookRequest(t, stream, &bytes.Buffer{})
+
+			require.Equal(t, tc.expectedExitStatus, status)
+			assert.Equal(t, tc.expectedStderr, text.ChompBytes(stderr))
+		})
+	}
+}