commit: Add CheckObjectsExist RPC

When pushing commits to a repository, access checks are run. In order to
use the quarantine directory, we need a way to filter out revisions that
a repository already has in the case that a packfile sends over objects
that already exists on the server. In this case, we don't need to check
the access.

Add an RPC that when given a list of revisions, returns the ones that
already exist in the repository, and the ones that do not exist in the
repository.

Changelog: added

2 files changed, 231 insertions, 0 deletions

diff --git a/internal/gitaly/service/commit/check_objects_exist.go b/internal/gitaly/service/commit/check_objects_exist.go
new file mode 100644
index 000000000..baab66f51
--- /dev/null
+++ b/internal/gitaly/service/commit/check_objects_exist.go
@@ -0,0 +1,110 @@
+package commit
+
+import (
+	"context"
+	"io"
+
+	"gitlab.com/gitlab-org/gitaly/v14/internal/git"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/git/catfile"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/helper/chunk"
+	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"google.golang.org/protobuf/proto"
+)
+
+func (s *server) CheckObjectsExist(
+	stream gitalypb.CommitService_CheckObjectsExistServer,
+) error {
+	ctx := stream.Context()
+
+	request, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+
+	if err := validateCheckObjectsExistRequest(request); err != nil {
+		return err
+	}
+
+	objectInfoReader, err := s.catfileCache.ObjectInfoReader(
+		ctx,
+		s.localrepo(request.GetRepository()),
+	)
+	if err != nil {
+		return err
+	}
+
+	chunker := chunk.New(&checkObjectsExistSender{stream: stream})
+	for {
+		request, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF {
+				return chunker.Flush()
+			}
+			return err
+		}
+
+		if err = checkObjectsExist(ctx, request, objectInfoReader, chunker); err != nil {
+			return err
+		}
+	}
+}
+
+type checkObjectsExistSender struct {
+	stream    gitalypb.CommitService_CheckObjectsExistServer
+	revisions []*gitalypb.CheckObjectsExistResponse_RevisionExistence
+}
+
+func (c *checkObjectsExistSender) Send() error {
+	return c.stream.Send(&gitalypb.CheckObjectsExistResponse{
+		Revisions: c.revisions,
+	})
+}
+
+func (c *checkObjectsExistSender) Reset() {
+	c.revisions = make([]*gitalypb.CheckObjectsExistResponse_RevisionExistence, 0)
+}
+
+func (c *checkObjectsExistSender) Append(m proto.Message) {
+	c.revisions = append(c.revisions, m.(*gitalypb.CheckObjectsExistResponse_RevisionExistence))
+}
+
+func checkObjectsExist(
+	ctx context.Context,
+	request *gitalypb.CheckObjectsExistRequest,
+	objectInfoReader catfile.ObjectInfoReader,
+	chunker *chunk.Chunker,
+) error {
+	revisions := request.GetRevisions()
+
+	for _, revision := range revisions {
+		revisionExistence := gitalypb.CheckObjectsExistResponse_RevisionExistence{
+			Name:   revision,
+			Exists: true,
+		}
+		_, err := objectInfoReader.Info(ctx, git.Revision(revision))
+		if err != nil {
+			if catfile.IsNotFound(err) {
+				revisionExistence.Exists = false
+			} else {
+				return err
+			}
+		}
+
+		if err := chunker.Send(&revisionExistence); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func validateCheckObjectsExistRequest(in *gitalypb.CheckObjectsExistRequest) error {
+	for _, revision := range in.GetRevisions() {
+		if err := git.ValidateRevision(revision); err != nil {
+			return helper.ErrInvalidArgument(err)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/gitaly/service/commit/check_objects_exist_test.go b/internal/gitaly/service/commit/check_objects_exist_test.go
new file mode 100644
index 000000000..97fed524f
--- /dev/null
+++ b/internal/gitaly/service/commit/check_objects_exist_test.go
@@ -0,0 +1,121 @@
+package commit
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/git/gittest"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/testhelper"
+	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"google.golang.org/grpc/codes"
+)
+
+func TestCheckObjectsExist(t *testing.T) {
+	t.Parallel()
+
+	ctx := testhelper.Context(t)
+	cfg, repo, repoPath, client := setupCommitServiceWithRepo(ctx, t, true)
+
+	// write a few commitIDs we can use
+	commitID1 := gittest.WriteCommit(t, cfg, repoPath)
+	commitID2 := gittest.WriteCommit(t, cfg, repoPath)
+	commitID3 := gittest.WriteCommit(t, cfg, repoPath)
+
+	// remove a ref from the repository so we know it doesn't exist
+	gittest.Exec(t, cfg, "-C", repoPath, "update-ref", "-d", "refs/heads/many_files")
+
+	nonexistingObject := "abcdefg"
+	cmd := gittest.NewCommand(t, cfg, "-C", repoPath, "rev-parse", nonexistingObject)
+	require.Error(t, cmd.Wait(), "ensure the object doesn't exist")
+
+	testCases := []struct {
+		desc               string
+		input              [][]byte
+		revisionsExistence map[string]bool
+		returnCode         codes.Code
+	}{
+		{
+			desc: "commit ids and refs that exist",
+			input: [][]byte{
+				[]byte(commitID1),
+				[]byte("master"),
+				[]byte(commitID2),
+				[]byte(commitID3),
+				[]byte("feature"),
+			},
+			revisionsExistence: map[string]bool{
+				"master":           true,
+				commitID2.String(): true,
+				commitID3.String(): true,
+				"feature":          true,
+			},
+			returnCode: codes.OK,
+		},
+		{
+			desc: "ref and objects missing",
+			input: [][]byte{
+				[]byte(commitID1),
+				[]byte("master"),
+				[]byte(commitID2),
+				[]byte(commitID3),
+				[]byte("feature"),
+				[]byte("many_files"),
+				[]byte(nonexistingObject),
+			},
+			revisionsExistence: map[string]bool{
+				"master":           true,
+				commitID2.String(): true,
+				commitID3.String(): true,
+				"feature":          true,
+				"many_files":       false,
+				nonexistingObject:  false,
+			},
+			returnCode: codes.OK,
+		},
+		{
+			desc:               "empty input",
+			input:              [][]byte{},
+			returnCode:         codes.OK,
+			revisionsExistence: map[string]bool{},
+		},
+		{
+			desc:       "invalid input",
+			input:      [][]byte{[]byte("-not-a-rev")},
+			returnCode: codes.InvalidArgument,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			c, err := client.CheckObjectsExist(ctx)
+			require.NoError(t, err)
+
+			require.NoError(t, c.Send(
+				&gitalypb.CheckObjectsExistRequest{
+					Repository: repo,
+					Revisions:  tc.input,
+				},
+			))
+			require.NoError(t, c.CloseSend())
+
+			for {
+				resp, err := c.Recv()
+				if tc.returnCode != codes.OK {
+					testhelper.RequireGrpcCode(t, err, tc.returnCode)
+					break
+				} else if err != nil {
+					require.Error(t, err, io.EOF)
+					break
+				}
+
+				actualRevisionsExistence := make(map[string]bool)
+				for _, revisionExistence := range resp.GetRevisions() {
+					actualRevisionsExistence[string(revisionExistence.GetName())] = revisionExistence.GetExists()
+				}
+				assert.Equal(t, tc.revisionsExistence, actualRevisionsExistence)
+			}
+		})
+	}
+}