Merge branch 'jt-atomic-hooks' into 'master'

hooks: Make RestoreCustomHooks RPC atomic

See merge request https://gitlab.com/gitlab-org/gitaly/-/merge_requests/5272

Merged-by: Justin Tobler <jtobler@gitlab.com>
Approved-by: Sami Hiltunen <shiltunen@gitlab.com>
Reviewed-by: James Fargher <proglottis@gmail.com>
Reviewed-by: Patrick Steinhardt <psteinhardt@gitlab.com>
Reviewed-by: Sami Hiltunen <shiltunen@gitlab.com>
Reviewed-by: Justin Tobler <jtobler@gitlab.com>

5 files changed, 127 insertions, 94 deletions

diff --git a/internal/gitaly/service/repository/restore_custom_hooks.go b/internal/gitaly/service/repository/restore_custom_hooks.go
index b221a32e5..51a82714b 100644
--- a/internal/gitaly/service/repository/restore_custom_hooks.go
+++ b/internal/gitaly/service/repository/restore_custom_hooks.go
@@ -12,29 +12,32 @@ import (
 
 	"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/command"
+	"gitlab.com/gitlab-org/gitaly/v15/internal/git/repository"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/gitaly/service"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/gitaly/transaction"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/safe"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/structerr"
+	"gitlab.com/gitlab-org/gitaly/v15/internal/tempdir"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/transaction/txinfo"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/transaction/voting"
 	"gitlab.com/gitlab-org/gitaly/v15/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/v15/streamio"
 )
 
+// RestoreCustomHooks sets the git hooks for a repository. The hooks are sent in
+// a tar archive containing a `custom_hooks` directory. This directory is
+// ultimately extracted to the repository.
 func (s *server) RestoreCustomHooks(stream gitalypb.RepositoryService_RestoreCustomHooksServer) error {
-	if featureflag.TransactionalRestoreCustomHooks.IsEnabled(stream.Context()) {
-		return s.restoreCustomHooksWithVoting(stream)
-	}
+	ctx := stream.Context()
 
 	firstRequest, err := stream.Recv()
 	if err != nil {
 		return structerr.NewInternal("first request failed %w", err)
 	}
 
-	repository := firstRequest.GetRepository()
-	if err := service.ValidateRepository(repository); err != nil {
+	repo := firstRequest.GetRepository()
+	if err := service.ValidateRepository(repo); err != nil {
 		return structerr.NewInvalidArgument("%w", err)
 	}
 
@@ -49,122 +52,119 @@ func (s *server) RestoreCustomHooks(stream gitalypb.RepositoryService_RestoreCus
 		return request.GetData(), err
 	})
 
-	repoPath, err := s.locator.GetPath(repository)
-	if err != nil {
-		return structerr.NewInternal("getting repo path failed %w", err)
-	}
+	if featureflag.TransactionalRestoreCustomHooks.IsEnabled(ctx) {
+		if err := s.restoreCustomHooks(ctx, reader, repo); err != nil {
+			return structerr.NewInternal("setting custom hooks: %w", err)
+		}
 
-	cmdArgs := []string{
-		"-xf",
-		"-",
-		"-C",
-		repoPath,
-		customHooksDir,
+		return stream.SendAndClose(&gitalypb.RestoreCustomHooksResponse{})
 	}
 
-	ctx := stream.Context()
-	cmd, err := command.New(ctx, append([]string{"tar"}, cmdArgs...), command.WithStdin(reader))
+	repoPath, err := s.locator.GetPath(repo)
 	if err != nil {
-		return structerr.NewInternal("Could not untar custom hooks tar %w", err)
+		return structerr.NewInternal("getting repo path failed %w", err)
 	}
 
-	if err := cmd.Wait(); err != nil {
-		return structerr.NewInternal("cmd wait failed: %w", err)
+	if err := extractHooks(ctx, reader, repoPath); err != nil {
+		return structerr.NewInternal("extracting hooks: %w", err)
 	}
 
 	return stream.SendAndClose(&gitalypb.RestoreCustomHooksResponse{})
 }
 
-func (s *server) restoreCustomHooksWithVoting(stream gitalypb.RepositoryService_RestoreCustomHooksServer) error {
-	firstRequest, err := stream.Recv()
+// restoreCustomHooks transactionally and atomically sets the provided custom
+// hooks for the specified repository.
+func (s *server) restoreCustomHooks(ctx context.Context, tar io.Reader, repo repository.GitRepo) error {
+	repoPath, err := s.locator.GetRepoPath(repo)
 	if err != nil {
-		return structerr.NewInternal("first request failed %w", err)
-	}
-
-	ctx := stream.Context()
-
-	repository := firstRequest.GetRepository()
-	if err := service.ValidateRepository(repository); err != nil {
-		return structerr.NewInvalidArgument("%w", err)
+		return fmt.Errorf("getting repo path: %w", err)
 	}
 
-	repoPath, err := s.locator.GetRepoPath(repository)
+	// The `custom_hooks` directory in the repository is locked to prevent
+	// concurrent modification of hooks.
+	hooksLock, err := safe.NewLockingDirectory(repoPath, customHooksDir)
 	if err != nil {
-		return structerr.NewInternal("RestoreCustomHooks: getting repo path failed %w", err)
+		return fmt.Errorf("creating hooks lock: %w", err)
 	}
 
-	customHooksPath := filepath.Join(repoPath, customHooksDir)
-
-	if err = os.MkdirAll(customHooksPath, os.ModePerm); err != nil {
-		return structerr.NewInternal("making custom hooks directory %w", err)
+	if err := hooksLock.Lock(); err != nil {
+		return fmt.Errorf("locking hooks: %w", err)
 	}
+	defer func() {
+		// If the `.lock` file is not removed from the `custom_hooks` directory,
+		// future modifications to the repository's hooks will be prevented. If
+		// this occurs, the `.lock` file will have to be manually removed.
+		if err := hooksLock.Unlock(); err != nil {
+			ctxlogrus.Extract(ctx).WithError(err).Error("failed to unlock hooks")
+		}
+	}()
 
-	lockDir, err := safe.NewLockingDirectory(repoPath, customHooksDir)
+	// Create a temporary directory to write the new hooks to and also
+	// temporarily store the current repository hooks. This enables "atomic"
+	// directory swapping by acting as an intermediary storage location between
+	// moves.
+	tmpDir, err := tempdir.NewWithoutContext(repo.GetStorageName(), s.locator)
 	if err != nil {
-		return structerr.NewInternal("RestoreCustomHooks: creating locking directory: %w", err)
-	}
-
-	if err := lockDir.Lock(); err != nil {
-		return structerr.NewInternal("locking directory failed: %w", err)
+		return fmt.Errorf("creating temp directory: %w", err)
 	}
 
 	defer func() {
-		if !lockDir.IsLocked() {
-			return
-		}
-
-		if err := lockDir.Unlock(); err != nil {
-			ctxlogrus.Extract(ctx).WithError(err).Warn("could not unlock directory")
+		if err := os.RemoveAll(tmpDir.Path()); err != nil {
+			ctxlogrus.Extract(ctx).WithError(err).Warn("failed to remove temporary directory")
 		}
 	}()
 
-	preparedVote := voting.NewVoteHash()
-	if err := voteCustomHooks(ctx, s.txManager, &preparedVote, voting.Prepared); err != nil {
-		return structerr.NewInternal("casting prepared vote: %w", err)
+	if err := extractHooks(ctx, tar, tmpDir.Path()); err != nil {
+		return fmt.Errorf("extracting hooks: %w", err)
 	}
 
-	reader := streamio.NewReader(func() ([]byte, error) {
-		if firstRequest != nil {
-			data := firstRequest.GetData()
-			firstRequest = nil
-			return data, nil
-		}
-
-		request, err := stream.Recv()
-		return request.GetData(), err
-	})
+	tempHooksPath := filepath.Join(tmpDir.Path(), customHooksDir)
 
-	cmdArgs := []string{
-		"-xf",
-		"-",
-		"-C",
-		repoPath,
-		customHooksDir,
+	// No hooks will be extracted if the tar archive is empty. If this happens
+	// it means the repository should be set with an empty `custom_hooks`
+	// directory. Create `custom_hooks` in the temporary directory so that any
+	// existing repository hooks will be replaced with this empty directory.
+	if err := os.Mkdir(tempHooksPath, os.ModePerm); err != nil && !errors.Is(err, fs.ErrExist) {
+		return fmt.Errorf("making temp hooks directory: %w", err)
 	}
 
-	cmd, err := command.New(ctx, append([]string{"tar"}, cmdArgs...), command.WithStdin(reader))
+	preparedVote, err := newDirectoryVote(tempHooksPath)
 	if err != nil {
-		return structerr.NewInternal("Could not untar custom hooks tar %w", err)
+		return fmt.Errorf("generating prepared vote: %w", err)
 	}
 
-	if err := cmd.Wait(); err != nil {
-		return structerr.NewInternal("cmd wait failed: %w", err)
+	// Cast prepared vote with hash of the extracted archive in the temporary
+	// `custom_hooks` directory.
+	if err := voteCustomHooks(ctx, s.txManager, preparedVote, voting.Prepared); err != nil {
+		return fmt.Errorf("casting prepared vote: %w", err)
 	}
 
-	committedVote, err := newDirectoryVote(customHooksPath)
-	if err != nil {
-		return structerr.NewInternal("generating committed vote: %w", err)
+	repoHooksPath := filepath.Join(repoPath, customHooksDir)
+	prevHooksPath := filepath.Join(tmpDir.Path(), "previous_hooks")
+
+	// If the `custom_hooks` directory exists in the repository, move the
+	// current hooks to `previous_hooks` in the temporary directory.
+	if err := os.Rename(repoHooksPath, prevHooksPath); err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return fmt.Errorf("moving current hooks to temp: %w", err)
 	}
 
-	if err := voteCustomHooks(ctx, s.txManager, committedVote, voting.Committed); err != nil {
-		return structerr.NewInternal("casting committed vote: %w", err)
+	// Move `custom_hooks` from the temporary directory to the repository.
+	if err := os.Rename(tempHooksPath, repoHooksPath); err != nil {
+		return fmt.Errorf("moving new hooks to repo: %w", err)
 	}
 
-	if err := lockDir.Unlock(); err != nil {
-		return structerr.NewInternal("committing lock dir %w", err)
+	committedVote, err := newDirectoryVote(repoHooksPath)
+	if err != nil {
+		return fmt.Errorf("generating committed vote: %w", err)
 	}
 
-	return stream.SendAndClose(&gitalypb.RestoreCustomHooksResponse{})
+	// Cast committed vote with hash of the extracted archive in the repository
+	// `custom_hooks` directory.
+	if err := voteCustomHooks(ctx, s.txManager, committedVote, voting.Committed); err != nil {
+		return fmt.Errorf("casting committed vote: %w", err)
+	}
+
+	return nil
 }
 
 // newDirectoryVote creates a voting.VoteHash by walking the specified path and
@@ -177,11 +177,13 @@ func newDirectoryVote(basePath string) (*voting.VoteHash, error) {
 			return err
 		}
 
-		// Write file name to hash. Since `WalkDir()` output is deterministic
-		// based on lexical order, the path does not need to be included with
-		// the name written to the hash. Any change to the entry's path will
-		// result in a different hash due to the change in walked order.
-		_, _ = voteHash.Write([]byte(entry.Name()))
+		relPath, err := filepath.Rel(basePath, path)
+		if err != nil {
+			return fmt.Errorf("getting relative path: %w", err)
+		}
+
+		// Write file relative path to hash.
+		_, _ = voteHash.Write([]byte(relPath))
 
 		info, err := entry.Info()
 		if err != nil {
@@ -216,6 +218,8 @@ func newDirectoryVote(basePath string) (*voting.VoteHash, error) {
 	return &voteHash, nil
 }
 
+// voteCustomHooks casts a vote symbolic of the custom hooks received. If there
+// is no transaction voting is skipped.
 func voteCustomHooks(
 	ctx context.Context,
 	txManager transaction.Manager,
@@ -240,3 +244,20 @@ func voteCustomHooks(
 
 	return nil
 }
+
+// extractHooks unpacks a tar file containing custom hooks into a `custom_hooks`
+// directory at the specified path.
+func extractHooks(ctx context.Context, reader io.Reader, path string) error {
+	cmdArgs := []string{"-xf", "-", "-C", path, customHooksDir}
+
+	cmd, err := command.New(ctx, append([]string{"tar"}, cmdArgs...), command.WithStdin(reader))
+	if err != nil {
+		return fmt.Errorf("executing tar command: %w", err)
+	}
+
+	if err := cmd.Wait(); err != nil {
+		return fmt.Errorf("waiting for tar command completion: %w", err)
+	}
+
+	return nil
+}
diff --git a/internal/gitaly/service/repository/restore_custom_hooks_test.go b/internal/gitaly/service/repository/restore_custom_hooks_test.go
index 6596c0301..48afedb17 100644
--- a/internal/gitaly/service/repository/restore_custom_hooks_test.go
+++ b/internal/gitaly/service/repository/restore_custom_hooks_test.go
@@ -180,7 +180,7 @@ func TestNewDirectoryVote(t *testing.T) {
 				{name: "pre-commit.sample", content: "foo", mode: 0o755},
 				{name: "pre-push.sample", content: "bar", mode: 0o755},
 			},
-			expectedHash: "b8f99a3012ce12c1a5711e3b94d45b98878cc18d",
+			expectedHash: "8ca11991268de4c9278488a674fc1a88db449566",
 		},
 		{
 			desc: "generated hash matches with changed file name",
@@ -188,7 +188,7 @@ func TestNewDirectoryVote(t *testing.T) {
 				{name: "pre-commit.sample.diff", content: "foo", mode: 0o755},
 				{name: "pre-push.sample", content: "bar", mode: 0o755},
 			},
-			expectedHash: "f55ff88d7f84045fce970615ecd04c4fe9bf0a94",
+			expectedHash: "b5ed58ced84103da1ed9d7813a9e39b3b5daf7d7",
 		},
 		{
 			desc: "generated hash matches with changed file content",
@@ -196,7 +196,7 @@ func TestNewDirectoryVote(t *testing.T) {
 				{name: "pre-commit.sample", content: "foo", mode: 0o755},
 				{name: "pre-push.sample", content: "bar.diff", mode: 0o755},
 			},
-			expectedHash: "0c628c59e62c351069037aad858a6f900ef1de9b",
+			expectedHash: "178083848c8a08e36c4f86c2d318a84b0bb845f2",
 		},
 		{
 			desc: "generated hash matches with changed file mode",
@@ -204,7 +204,7 @@ func TestNewDirectoryVote(t *testing.T) {
 				{name: "pre-commit.sample", content: "foo", mode: 0o644},
 				{name: "pre-push.sample", content: "bar", mode: 0o755},
 			},
-			expectedHash: "9321dde590ef5ffa05d1ddf690b8983f406242e7",
+			expectedHash: "c69574241b83496bb4005b4f7a0dfcda96cb317e",
 		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
diff --git a/internal/praefect/coordinator.go b/internal/praefect/coordinator.go
index 51ddd0b45..b64a827ca 100644
--- a/internal/praefect/coordinator.go
+++ b/internal/praefect/coordinator.go
@@ -42,6 +42,12 @@ type transactionsCondition func(context.Context) bool
 func transactionsEnabled(context.Context) bool  { return true }
 func transactionsDisabled(context.Context) bool { return false }
 
+func transactionsFlag(flag featureflag.FeatureFlag) transactionsCondition {
+	return func(ctx context.Context) bool {
+		return flag.IsEnabled(ctx)
+	}
+}
+
 // transactionRPCs contains the list of repository-scoped mutating calls which may take part in
 // transactions. An optional feature flag can be added to conditionally enable transactional
 // behaviour. If none is given, it's always enabled.
@@ -90,9 +96,9 @@ var transactionRPCs = map[string]transactionsCondition{
 	"/gitaly.ObjectPoolService/ReduplicateRepository":      transactionsDisabled,
 	"/gitaly.RepositoryService/RenameRepository":           transactionsDisabled,
 
-	// This RPC call should be made transactional. Furthermore, we should consider whether we
-	// have to replicate custom hooks.
-	"/gitaly.RepositoryService/RestoreCustomHooks": transactionsDisabled,
+	// The `RestoreCustomHooks` RPC can be make transactional by enabling the
+	// `TransactionalRestoreCustomHooks` feature flag.
+	"/gitaly.RepositoryService/RestoreCustomHooks": transactionsFlag(featureflag.TransactionalRestoreCustomHooks),
 }
 
 // forcePrimaryRoutingRPCs tracks RPCs which need to always get routed to the primary. This should
diff --git a/proto/go/gitalypb/repository_grpc.pb.go b/proto/go/gitalypb/repository_grpc.pb.go
index 07cec846d..58a7761e1 100644
--- a/proto/go/gitalypb/repository_grpc.pb.go
+++ b/proto/go/gitalypb/repository_grpc.pb.go
@@ -100,7 +100,9 @@ type RepositoryServiceClient interface {
 	SearchFilesByContent(ctx context.Context, in *SearchFilesByContentRequest, opts ...grpc.CallOption) (RepositoryService_SearchFilesByContentClient, error)
 	// This comment is left unintentionally blank.
 	SearchFilesByName(ctx context.Context, in *SearchFilesByNameRequest, opts ...grpc.CallOption) (RepositoryService_SearchFilesByNameClient, error)
-	// This comment is left unintentionally blank.
+	// RestoreCustomHooks sets the git hooks for a repository. The hooks are sent
+	// in a tar archive containing a `custom_hooks` directory. This directory is
+	// ultimately extracted to the repository.
 	RestoreCustomHooks(ctx context.Context, opts ...grpc.CallOption) (RepositoryService_RestoreCustomHooksClient, error)
 	// This comment is left unintentionally blank.
 	BackupCustomHooks(ctx context.Context, in *BackupCustomHooksRequest, opts ...grpc.CallOption) (RepositoryService_BackupCustomHooksClient, error)
@@ -925,7 +927,9 @@ type RepositoryServiceServer interface {
 	SearchFilesByContent(*SearchFilesByContentRequest, RepositoryService_SearchFilesByContentServer) error
 	// This comment is left unintentionally blank.
 	SearchFilesByName(*SearchFilesByNameRequest, RepositoryService_SearchFilesByNameServer) error
-	// This comment is left unintentionally blank.
+	// RestoreCustomHooks sets the git hooks for a repository. The hooks are sent
+	// in a tar archive containing a `custom_hooks` directory. This directory is
+	// ultimately extracted to the repository.
 	RestoreCustomHooks(RepositoryService_RestoreCustomHooksServer) error
 	// This comment is left unintentionally blank.
 	BackupCustomHooks(*BackupCustomHooksRequest, RepositoryService_BackupCustomHooksServer) error
diff --git a/proto/repository.proto b/proto/repository.proto
index 048738b15..2b8e07dbe 100644
--- a/proto/repository.proto
+++ b/proto/repository.proto
@@ -248,7 +248,9 @@ service RepositoryService {
     };
   }
 
-  // This comment is left unintentionally blank.
+  // RestoreCustomHooks sets the git hooks for a repository. The hooks are sent
+  // in a tar archive containing a `custom_hooks` directory. This directory is
+  // ultimately extracted to the repository.
   rpc RestoreCustomHooks(stream RestoreCustomHooksRequest) returns (RestoreCustomHooksResponse) {
     option (op_type) = {
       op: MUTATOR