Implement BlobService.GetNewLfsPointers

10 files changed, 344 insertions, 69 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cecc1a2bd..f5e92f69b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 UNRELEASED
 
+- Implement BlobService.GetNewLfsPointers
+  https://gitlab.com/gitlab-org/gitaly/merge_requests/562
 - Use gitaly-proto v0.86.0
   https://gitlab.com/gitlab-org/gitaly/merge_requests/606
 
diff --git a/internal/service/blob/lfs_pointers.go b/internal/service/blob/lfs_pointers.go
index 74c855819..c0104e527 100644
--- a/internal/service/blob/lfs_pointers.go
+++ b/internal/service/blob/lfs_pointers.go
@@ -3,6 +3,7 @@ package blob
 import (
 	"fmt"
 
+	"gitlab.com/gitlab-org/gitaly/internal/git"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/rubyserver"
 
@@ -57,8 +58,45 @@ func validateGetLFSPointersRequest(req *pb.GetLFSPointersRequest) error {
 	return nil
 }
 
-func (*server) GetNewLFSPointers(*pb.GetNewLFSPointersRequest, pb.BlobService_GetNewLFSPointersServer) error {
-	return helper.Unimplemented
+func (s *server) GetNewLFSPointers(in *pb.GetNewLFSPointersRequest, stream pb.BlobService_GetNewLFSPointersServer) error {
+	ctx := stream.Context()
+
+	if err := validateGetNewLFSPointersRequest(in); err != nil {
+		return status.Errorf(codes.InvalidArgument, "GetNewLFSPointers: %v", err)
+	}
+
+	client, err := s.BlobServiceClient(ctx)
+	if err != nil {
+		return err
+	}
+
+	clientCtx, err := rubyserver.SetHeaders(ctx, in.GetRepository())
+	if err != nil {
+		return err
+	}
+
+	rubyStream, err := client.GetNewLFSPointers(clientCtx, in)
+	if err != nil {
+		return err
+	}
+
+	return rubyserver.Proxy(func() error {
+		resp, err := rubyStream.Recv()
+		if err != nil {
+			md := rubyStream.Trailer()
+			stream.SetTrailer(md)
+			return err
+		}
+		return stream.Send(resp)
+	})
+}
+
+func validateGetNewLFSPointersRequest(in *pb.GetNewLFSPointersRequest) error {
+	if in.GetRepository() == nil {
+		return fmt.Errorf("empty Repository")
+	}
+
+	return git.ValidateRevision(in.GetRevision())
 }
 
 func (*server) GetAllLFSPointers(*pb.GetAllLFSPointersRequest, pb.BlobService_GetAllLFSPointersServer) error {
diff --git a/internal/service/blob/lfs_pointers_test.go b/internal/service/blob/lfs_pointers_test.go
index 8320edaf2..6f5becd5a 100644
--- a/internal/service/blob/lfs_pointers_test.go
+++ b/internal/service/blob/lfs_pointers_test.go
@@ -2,6 +2,7 @@ package blob
 
 import (
 	"io"
+	"os/exec"
 	"testing"
 
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
@@ -123,3 +124,224 @@ func TestFailedGetLFSPointersRequestDueToValidations(t *testing.T) {
 		})
 	}
 }
+
+func TestSuccessfulGetNewLFSPointersRequest(t *testing.T) {
+	server, serverSocketPath := runBlobServer(t)
+	defer server.Stop()
+
+	client, conn := newBlobClient(t, serverSocketPath)
+	defer conn.Close()
+
+	testRepo, testRepoPath, cleanupFn := testhelper.NewTestRepoWithWorktree(t)
+	defer cleanupFn()
+
+	revision := []byte("46abbb087fcc0fd02c340f0f2f052bd2c7708da3")
+	cmd := exec.Command("git", "-C", testRepoPath, "cherry-pick", string(revision))
+	altDirsCommit, altDirs := testhelper.CreateCommitInAlternateObjectDirectory(t, testRepoPath, cmd)
+
+	// Create a commit not pointed at by any ref to emulate being in the
+	// pre-receive hook so that `--not --all` returns some objects
+	newRevision := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoPath, "commit-tree", "8856a329dd38ca86dfb9ce5aa58a16d88cc119bd", "-m", "Add LFS objects")
+	newRevision = newRevision[:len(newRevision)-1] // Strip newline
+
+	testCases := []struct {
+		desc                string
+		request             *pb.GetNewLFSPointersRequest
+		expectedLFSPointers []*pb.LFSPointer
+	}{
+		{
+			desc: "standard request",
+			request: &pb.GetNewLFSPointersRequest{
+				Repository: testRepo,
+				Revision:   revision,
+			},
+			expectedLFSPointers: []*pb.LFSPointer{
+				{
+					Size: 133,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:91eff75a492a3ed0dfcb544d7f31326bc4014c8551849c192fd1e48d4dd2c897\nsize 1575078\n\n"),
+					Oid:  "0c304a93cb8430108629bbbcaa27db3343299bc0",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:f2b0a1e7550e9b718dafc9b525a04879a766de62e4fbdfc46593d47f7ab74636\nsize 20\n"),
+					Oid:  "f78df813119a79bfbe0442ab92540a61d3ab7ff3",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:bad71f905b60729f502ca339f7c9f001281a3d12c68a5da7f15de8009f4bd63d\nsize 18\n"),
+					Oid:  "bab31d249f78fba464d1b75799aad496cc07fa3b",
+				},
+			},
+		},
+		{
+			desc: "request with revision in alternate directory",
+			request: &pb.GetNewLFSPointersRequest{
+				Repository: testRepo,
+				Revision:   altDirsCommit,
+			},
+			expectedLFSPointers: []*pb.LFSPointer{
+				{
+					Size: 133,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:91eff75a492a3ed0dfcb544d7f31326bc4014c8551849c192fd1e48d4dd2c897\nsize 1575078\n\n"),
+					Oid:  "0c304a93cb8430108629bbbcaa27db3343299bc0",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:f2b0a1e7550e9b718dafc9b525a04879a766de62e4fbdfc46593d47f7ab74636\nsize 20\n"),
+					Oid:  "f78df813119a79bfbe0442ab92540a61d3ab7ff3",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:bad71f905b60729f502ca339f7c9f001281a3d12c68a5da7f15de8009f4bd63d\nsize 18\n"),
+					Oid:  "bab31d249f78fba464d1b75799aad496cc07fa3b",
+				},
+			},
+		},
+		{
+			desc: "request with limit",
+			request: &pb.GetNewLFSPointersRequest{
+				Repository: testRepo,
+				Revision:   revision,
+				// This is limiting the amount of lines processed from the
+				// output of rev-list. For example, for this revision's  output
+				// there's an LFS object id in line 4 and another in line 14, so
+				// any limit in [0, 3] will yield no pointers, [4,13] will yield
+				// one, and [14,] will yield two. This is weird but it's the
+				// way the current implementation works ¯\_(ツ)_/¯
+				Limit: 19,
+			},
+			expectedLFSPointers: []*pb.LFSPointer{
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:bad71f905b60729f502ca339f7c9f001281a3d12c68a5da7f15de8009f4bd63d\nsize 18\n"),
+					Oid:  "bab31d249f78fba464d1b75799aad496cc07fa3b",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:f2b0a1e7550e9b718dafc9b525a04879a766de62e4fbdfc46593d47f7ab74636\nsize 20\n"),
+					Oid:  "f78df813119a79bfbe0442ab92540a61d3ab7ff3",
+				},
+			},
+		},
+		{
+			desc: "with NotInAll true",
+			request: &pb.GetNewLFSPointersRequest{
+				Repository: testRepo,
+				Revision:   newRevision,
+				NotInAll:   true,
+			},
+			expectedLFSPointers: []*pb.LFSPointer{
+				{
+					Size: 133,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:91eff75a492a3ed0dfcb544d7f31326bc4014c8551849c192fd1e48d4dd2c897\nsize 1575078\n\n"),
+					Oid:  "0c304a93cb8430108629bbbcaa27db3343299bc0",
+				},
+			},
+		},
+		{
+			desc: "with some NotInRefs elements",
+			request: &pb.GetNewLFSPointersRequest{
+				Repository: testRepo,
+				Revision:   revision,
+				NotInRefs:  [][]byte{[]byte("048721d90c449b244b7b4c53a9186b04330174ec")},
+			},
+			expectedLFSPointers: []*pb.LFSPointer{
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:bad71f905b60729f502ca339f7c9f001281a3d12c68a5da7f15de8009f4bd63d\nsize 18\n"),
+					Oid:  "bab31d249f78fba464d1b75799aad496cc07fa3b",
+				},
+				{
+					Size: 127,
+					Data: []byte("version https://git-lfs.github.com/spec/v1\noid sha256:f2b0a1e7550e9b718dafc9b525a04879a766de62e4fbdfc46593d47f7ab74636\nsize 20\n"),
+					Oid:  "f78df813119a79bfbe0442ab92540a61d3ab7ff3",
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			tc.request.Repository.GitAlternateObjectDirectories = []string{altDirs}
+			stream, err := client.GetNewLFSPointers(ctx, tc.request)
+			require.NoError(t, err)
+
+			var receivedLFSPointers []*pb.LFSPointer
+			for {
+				resp, err := stream.Recv()
+				if err == io.EOF {
+					break
+				} else if err != nil {
+					t.Fatal(err)
+				}
+
+				receivedLFSPointers = append(receivedLFSPointers, resp.GetLfsPointers()...)
+			}
+
+			require.ElementsMatch(t, receivedLFSPointers, tc.expectedLFSPointers)
+		})
+	}
+}
+
+func TestFailedGetNewLFSPointersRequestDueToValidations(t *testing.T) {
+	server, serverSocketPath := runBlobServer(t)
+	defer server.Stop()
+
+	client, conn := newBlobClient(t, serverSocketPath)
+	defer conn.Close()
+
+	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+	defer cleanupFn()
+
+	testCases := []struct {
+		desc       string
+		repository *pb.Repository
+		revision   []byte
+	}{
+		{
+			desc:       "empty Repository",
+			repository: nil,
+			revision:   []byte("master"),
+		},
+		{
+			desc:       "empty revision",
+			repository: testRepo,
+			revision:   nil,
+		},
+		{
+			desc:       "revision can't start with '-'",
+			repository: testRepo,
+			revision:   []byte("-suspicious-revision"),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			request := &pb.GetNewLFSPointersRequest{
+				Repository: tc.repository,
+				Revision:   tc.revision,
+			}
+
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			c, err := client.GetNewLFSPointers(ctx, request)
+			require.NoError(t, err)
+
+			err = drainNewPointers(c)
+			testhelper.AssertGrpcError(t, err, codes.InvalidArgument, tc.desc)
+		})
+	}
+}
+
+func drainNewPointers(c pb.BlobService_GetNewLFSPointersClient) error {
+	for {
+		_, err := c.Recv()
+		if err != nil {
+			return err
+		}
+	}
+}
diff --git a/internal/service/commit/find_commits_test.go b/internal/service/commit/find_commits_test.go
index 1eb970c88..740935502 100644
--- a/internal/service/commit/find_commits_test.go
+++ b/internal/service/commit/find_commits_test.go
@@ -5,9 +5,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"os"
 	"os/exec"
-	"path"
 	"testing"
 
 	"google.golang.org/grpc/codes"
@@ -283,35 +281,11 @@ func TestSuccessfulFindCommitsRequestWithAltGitObjectDirs(t *testing.T) {
 	testRepoCopy, testRepoCopyPath, cleanupFn := testhelper.NewTestRepoWithWorktree(t)
 	defer cleanupFn()
 
-	altObjectsDir := "./alt-objects"
-	altObjectsPath := path.Join(testRepoCopyPath, ".git", altObjectsDir)
-	gitObjectEnv := []string{
-		fmt.Sprintf("GIT_OBJECT_DIRECTORY=%s", altObjectsPath),
-		fmt.Sprintf("GIT_ALTERNATE_OBJECT_DIRECTORIES=%s", path.Join(testRepoCopyPath, ".git/objects")),
-	}
-
-	if err := os.Mkdir(altObjectsPath, 0777); err != nil {
-		t.Fatal(err)
-	}
-
 	cmd := exec.Command("git", "-C", testRepoCopyPath,
 		"-c", fmt.Sprintf("user.name=%s", committerName),
 		"-c", fmt.Sprintf("user.email=%s", committerEmail),
 		"commit", "--allow-empty", "-m", "An empty commit")
-	// Because we set 'gitObjectEnv', the new objects created by this 'git commit' command will go
-	// into 'find-commits-alt-test-repo/.git/alt-objects'.
-	cmd.Env = gitObjectEnv
-	if _, err := cmd.Output(); err != nil {
-		stderr := err.(*exec.ExitError).Stderr
-		t.Fatalf("%s", stderr)
-	}
-
-	cmd = exec.Command("git", "-C", testRepoCopyPath, "show", "--format=format:%H", "--no-patch", "HEAD")
-	cmd.Env = gitObjectEnv
-	currentHead, err := cmd.Output()
-	if err != nil {
-		t.Fatal(err)
-	}
+	currentHead, altObjectsDir := testhelper.CreateCommitInAlternateObjectDirectory(t, testRepoCopyPath, cmd)
 
 	testCases := []struct {
 		desc          string
diff --git a/internal/service/commit/isancestor_test.go b/internal/service/commit/isancestor_test.go
index 1bd7c6c81..9a8e0530c 100644
--- a/internal/service/commit/isancestor_test.go
+++ b/internal/service/commit/isancestor_test.go
@@ -2,9 +2,7 @@ package commit
 
 import (
 	"fmt"
-	"os"
 	"os/exec"
-	"path"
 	"testing"
 
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
@@ -193,35 +191,13 @@ func TestSuccessfulIsAncestorRequestWithAltGitObjectDirs(t *testing.T) {
 	testRepoCopy, testRepoCopyPath, cleanupFn := testhelper.NewTestRepoWithWorktree(t)
 	defer cleanupFn()
 
-	altObjectsDir := "./alt-objects"
-	altObjectsPath := path.Join(testRepoCopyPath, ".git", altObjectsDir)
-	gitObjectEnv := []string{
-		fmt.Sprintf("GIT_OBJECT_DIRECTORY=%s", altObjectsPath),
-		fmt.Sprintf("GIT_ALTERNATE_OBJECT_DIRECTORIES=%s", path.Join(testRepoCopyPath, ".git/objects")),
-	}
-
-	if err := os.Mkdir(altObjectsPath, 0777); err != nil {
-		t.Fatal(err)
-	}
-
 	previousHead := testhelper.MustRunCommand(t, nil, "git", "-C", testRepoCopyPath, "show", "--format=format:%H", "--no-patch", "HEAD")
 
 	cmd := exec.Command("git", "-C", testRepoCopyPath,
 		"-c", fmt.Sprintf("user.name=%s", committerName),
 		"-c", fmt.Sprintf("user.email=%s", committerEmail),
 		"commit", "--allow-empty", "-m", "An empty commit")
-	cmd.Env = gitObjectEnv
-	if _, err := cmd.Output(); err != nil {
-		stderr := err.(*exec.ExitError).Stderr // XXX
-		t.Fatalf("%s", stderr)
-	}
-
-	cmd = exec.Command("git", "-C", testRepoCopyPath, "show", "--format=format:%H", "--no-patch", "HEAD")
-	cmd.Env = gitObjectEnv
-	currentHead, err := cmd.Output()
-	if err != nil {
-		t.Fatal(err)
-	}
+	currentHead, altObjectsDir := testhelper.CreateCommitInAlternateObjectDirectory(t, testRepoCopyPath, cmd)
 
 	testCases := []struct {
 		desc    string
diff --git a/internal/testhelper/commit.go b/internal/testhelper/commit.go
index 5d3cca8da..e0285b6b9 100644
--- a/internal/testhelper/commit.go
+++ b/internal/testhelper/commit.go
@@ -1,6 +1,10 @@
 package testhelper
 
 import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
 	"strings"
 	"testing"
 )
@@ -20,3 +24,37 @@ func CreateCommit(t *testing.T, repoPath string, branchName string) string {
 	MustRunCommand(t, nil, "git", "-C", repoPath, "update-ref", "refs/heads/"+branchName, newCommitID)
 	return newCommitID
 }
+
+// CreateCommitInAlternateObjectDirectory runs a command such that its created
+// objects will live in an alternate objects directory. It returns the current
+// head after the command is run and the alternate objects directory path
+func CreateCommitInAlternateObjectDirectory(t *testing.T, repoPath string, cmd *exec.Cmd) (currentHead []byte, altObjectsDir string) {
+	gitPath := path.Join(repoPath, ".git")
+	altObjectsDir = "./alt-objects"
+
+	altObjectsPath := path.Join(gitPath, altObjectsDir)
+	gitObjectEnv := []string{
+		fmt.Sprintf("GIT_OBJECT_DIRECTORY=%s", altObjectsPath),
+		fmt.Sprintf("GIT_ALTERNATE_OBJECT_DIRECTORIES=%s", path.Join(gitPath, "objects")),
+	}
+	if err := os.Mkdir(altObjectsPath, 0777); err != nil {
+		t.Fatal(err)
+	}
+
+	// Because we set 'gitObjectEnv', the new objects created by this command
+	// will go into 'find-commits-alt-test-repo/.git/alt-objects'.
+	cmd.Env = gitObjectEnv
+	if output, err := cmd.Output(); err != nil {
+		stderr := err.(*exec.ExitError).Stderr
+		t.Fatalf("stdout: %s, stderr: %s", output, stderr)
+	}
+
+	cmd = exec.Command("git", "-C", repoPath, "rev-parse", "HEAD")
+	cmd.Env = gitObjectEnv
+	currentHead, err := cmd.Output()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return currentHead[:len(currentHead)-1], altObjectsDir
+}
diff --git a/internal/testhelper/testhelper.go b/internal/testhelper/testhelper.go
index 52cea227c..980d1040f 100644
--- a/internal/testhelper/testhelper.go
+++ b/internal/testhelper/testhelper.go
@@ -139,7 +139,7 @@ func AssertGrpcError(t *testing.T, err error, expectedCode codes.Code, containsT
 	}
 
 	if containsText != "" && !strings.Contains(err.Error(), containsText) {
-		t.Fatal(err)
+		t.Fatalf("Expected an error message containing %v, got %v", containsText, err.Error())
 	}
 }
 
diff --git a/ruby/Gemfile b/ruby/Gemfile
index 74c464336..9744bbb65 100644
--- a/ruby/Gemfile
+++ b/ruby/Gemfile
@@ -2,7 +2,7 @@ source 'https://rubygems.org'
 
 gem 'github-linguist', '~> 4.7.0', require: 'linguist'
 gem 'gitlab-markup', '~> 1.6.2'
-gem 'gitaly-proto', '~> 0.85.0', require: 'gitaly'
+gem 'gitaly-proto', '~> 0.86.0', require: 'gitaly'
 gem 'activesupport', '~> 5.0.2'
 gem 'rdoc', '~> 4.2'
 gem 'gollum-lib', '~> 4.2', require: false
diff --git a/ruby/Gemfile.lock b/ruby/Gemfile.lock
index 8ad696dc6..bb0bbbce2 100644
--- a/ruby/Gemfile.lock
+++ b/ruby/Gemfile.lock
@@ -17,7 +17,7 @@ GEM
       multipart-post (>= 1.2, < 3)
     gemojione (3.3.0)
       json
-    gitaly-proto (0.85.0)
+    gitaly-proto (0.86.0)
       google-protobuf (~> 3.1)
       grpc (~> 1.0)
     github-linguist (4.7.6)
@@ -141,7 +141,7 @@ PLATFORMS
 
 DEPENDENCIES
   activesupport (~> 5.0.2)
-  gitaly-proto (~> 0.85.0)
+  gitaly-proto (~> 0.86.0)
   github-linguist (~> 4.7.0)
   gitlab-markup (~> 1.6.2)
   gitlab-styles (~> 2.0.0)
diff --git a/ruby/lib/gitaly_server/blob_service.rb b/ruby/lib/gitaly_server/blob_service.rb
index f1b336edc..cb9fb249d 100644
--- a/ruby/lib/gitaly_server/blob_service.rb
+++ b/ruby/lib/gitaly_server/blob_service.rb
@@ -2,27 +2,52 @@ module GitalyServer
   class BlobService < Gitaly::BlobService::Service
     include Utils
 
+    # LFS pointers are maximum 200 bytes in size, and for a default message size
+    # of 4 MB we got more than enough room for 100 blobs.
+    MAX_LFS_POINTERS_PER_MESSSAGE = 100
+
     def get_lfs_pointers(request, call)
       bridge_exceptions do
         repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
         blobs = Gitlab::Git::Blob.batch_lfs_pointers(repo, request.blob_ids)
 
         Enumerator.new do |y|
-          # LFS pointers are maximum 200 bytes in size, and for a default message size of 4 MB
-          # we got more than enough room for 100 blobs.
-          blobs.each_slice(100) do |blobs_slice|
-            lfs_pointers = blobs_slice.map do |blob|
-              Gitaly::LFSPointer.new(
-                size: blob.size,
-                data: blob.data,
-                oid: blob.id
-              )
-            end
-
+          sliced_gitaly_lfs_pointers(blobs) do |lfs_pointers|
             y.yield Gitaly::GetLFSPointersResponse.new(lfs_pointers: lfs_pointers)
           end
         end
       end
     end
+
+    def get_new_lfs_pointers(request, call)
+      Enumerator.new do |y|
+        bridge_exceptions do
+          changes = lfs_changes(request, call)
+          object_limit = request.limit.zero? ? nil : request.limit
+          not_in = request.not_in_all ? :all : request.not_in_refs.to_a
+          blobs = changes.new_pointers(object_limit: object_limit, not_in: not_in)
+
+          sliced_gitaly_lfs_pointers(blobs) do |lfs_pointers|
+            y.yield Gitaly::GetNewLFSPointersResponse.new(lfs_pointers: lfs_pointers)
+          end
+        end
+      end
+    end
+
+    private
+
+    def lfs_changes(request, call)
+      repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
+
+      Gitlab::Git::LfsChanges.new(repo, request.revision)
+    end
+
+    def sliced_gitaly_lfs_pointers(blobs)
+      blobs.each_slice(MAX_LFS_POINTERS_PER_MESSSAGE) do |blobs_slice|
+        yield (blobs_slice.map do |blob|
+          Gitaly::LFSPointer.new(size: blob.size, data: blob.data, oid: blob.id)
+        end)
+      end
+    end
   end
 end