Merge branch 'smh-fix-dir-deleted-walk' into 'master'

Ignore repositories deleted during a file walk

Closes #2611

See merge request gitlab-org/gitaly!2083

5 files changed, 98 insertions, 35 deletions

diff --git a/changelogs/unreleased/smh-fix-dir-deleted-walk.yml b/changelogs/unreleased/smh-fix-dir-deleted-walk.yml
new file mode 100644
index 000000000..a2d5f2c43
--- /dev/null
+++ b/changelogs/unreleased/smh-fix-dir-deleted-walk.yml
@@ -0,0 +1,5 @@
+---
+title: Ignore repositories deleted during a file walk
+merge_request: 2083
+author:
+type: fixed
diff --git a/internal/service/internalgitaly/testhelper_test.go b/internal/service/internalgitaly/testhelper_test.go
index 54ec265e6..bccec58af 100644
--- a/internal/service/internalgitaly/testhelper_test.go
+++ b/internal/service/internalgitaly/testhelper_test.go
@@ -2,22 +2,15 @@ package internalgitaly
 
 import (
 	"net"
-	"os"
 	"testing"
 
-	"gitlab.com/gitlab-org/gitaly/internal/config"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
 )
 
-func TestMain(m *testing.M) {
-	testhelper.Configure()
-	os.Exit(m.Run())
-}
-
-func runInternalGitalyServer(t *testing.T) (*grpc.Server, string) {
+func runInternalGitalyServer(t *testing.T, srv gitalypb.InternalGitalyServer) (*grpc.Server, string) {
 	serverSocketPath := testhelper.GetTemporaryGitalySocketFileName()
 	grpcServer := testhelper.NewTestGrpcServer(t, nil, nil)
 
@@ -26,7 +19,7 @@ func runInternalGitalyServer(t *testing.T) (*grpc.Server, string) {
 		t.Fatal(err)
 	}
 
-	gitalypb.RegisterInternalGitalyServer(grpcServer, NewServer(config.Config.Storages))
+	gitalypb.RegisterInternalGitalyServer(grpcServer, srv)
 	reflection.Register(grpcServer)
 
 	go grpcServer.Serve(listener)
diff --git a/internal/service/internalgitaly/walkrepos.go b/internal/service/internalgitaly/walkrepos.go
index 4f167094c..8691634b6 100644
--- a/internal/service/internalgitaly/walkrepos.go
+++ b/internal/service/internalgitaly/walkrepos.go
@@ -35,6 +35,10 @@ func (s *server) storagePath(storageName string) (string, error) {
 func walkStorage(ctx context.Context, storagePath string, stream gitalypb.InternalGitaly_WalkReposServer) error {
 	return filepath.Walk(storagePath, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
+			if os.IsNotExist(err) {
+				return nil
+			}
+
 			return err
 		}
 
diff --git a/internal/service/internalgitaly/walkrepos_test.go b/internal/service/internalgitaly/walkrepos_test.go
index 318720334..c5b2a1ff5 100644
--- a/internal/service/internalgitaly/walkrepos_test.go
+++ b/internal/service/internalgitaly/walkrepos_test.go
@@ -2,6 +2,9 @@ package internalgitaly
 
 import (
 	"io"
+	"os"
+	"path/filepath"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -12,19 +15,62 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+type serverWrapper struct {
+	gitalypb.InternalGitalyServer
+	WalkReposFunc func(*gitalypb.WalkReposRequest, gitalypb.InternalGitaly_WalkReposServer) error
+}
+
+func (w *serverWrapper) WalkRepos(req *gitalypb.WalkReposRequest, stream gitalypb.InternalGitaly_WalkReposServer) error {
+	return w.WalkReposFunc(req, stream)
+}
+
+type streamWrapper struct {
+	gitalypb.InternalGitaly_WalkReposServer
+	SendFunc func(*gitalypb.WalkReposResponse) error
+}
+
+func (w *streamWrapper) Send(resp *gitalypb.WalkReposResponse) error {
+	return w.SendFunc(resp)
+}
+
 func TestWalkRepos(t *testing.T) {
-	server, serverSocketPath := runInternalGitalyServer(t)
+	testRoot, clean := testhelper.TempDir(t)
+	defer clean()
+
+	storageName := "default"
+	storageRoot := filepath.Join(testRoot, "storage")
+
+	// file walk happens lexicographically, so we delete repository in the middle
+	// of the seqeuence to ensure the walk proceeds normally
+	testRepo1 := testhelper.NewTestRepoTo(t, storageRoot, "a")
+	deletedRepo := testhelper.NewTestRepoTo(t, storageRoot, "b")
+	testRepo2 := testhelper.NewTestRepoTo(t, storageRoot, "c")
+
+	// to test a directory being deleted during a walk, we must delete a directory after
+	// the file walk has started. To achieve that, we wrap the server to pass down a wrapped
+	// stream that allows us to hook in to stream responses. We then delete 'b' when
+	// the first repo 'a' is being streamed to the client.
+	deleteOnce := sync.Once{}
+	srv := NewServer([]config.Storage{{Name: storageName, Path: storageRoot}})
+	wsrv := &serverWrapper{srv,
+		func(r *gitalypb.WalkReposRequest, s gitalypb.InternalGitaly_WalkReposServer) error {
+			return srv.WalkRepos(r, &streamWrapper{s,
+				func(resp *gitalypb.WalkReposResponse) error {
+					deleteOnce.Do(func() {
+						require.NoError(t, os.RemoveAll(filepath.Join(storageRoot, deletedRepo.RelativePath)))
+					})
+					return s.Send(resp)
+				},
+			})
+		},
+	}
+
+	server, serverSocketPath := runInternalGitalyServer(t, wsrv)
 	defer server.Stop()
 
 	client, conn := newInternalGitalyClient(t, serverSocketPath)
 	defer conn.Close()
 
-	testRepo1, _, cleanupFn1 := testhelper.NewTestRepo(t)
-	defer cleanupFn1()
-
-	testRepo2, _, cleanupFn2 := testhelper.NewTestRepo(t)
-	defer cleanupFn2()
-
 	ctx, cancel := testhelper.Context()
 	defer cancel()
 
@@ -40,13 +86,15 @@ func TestWalkRepos(t *testing.T) {
 	require.Equal(t, codes.NotFound, s.Code())
 
 	stream, err = client.WalkRepos(ctx, &gitalypb.WalkReposRequest{
-		StorageName: config.Config.Storages[0].Name,
+		StorageName: storageName,
 	})
 	require.NoError(t, err)
 
 	actualRepos := consumeWalkReposStream(t, stream)
-	require.Contains(t, actualRepos, testRepo1.GetRelativePath())
-	require.Contains(t, actualRepos, testRepo2.GetRelativePath())
+	require.Equal(t, []string{
+		testRepo1.GetRelativePath(),
+		testRepo2.GetRelativePath(),
+	}, actualRepos)
 }
 
 func consumeWalkReposStream(t *testing.T, stream gitalypb.InternalGitaly_WalkReposClient) []string {
diff --git a/internal/testhelper/testhelper.go b/internal/testhelper/testhelper.go
index 03712db58..208c28f2d 100644
--- a/internal/testhelper/testhelper.go
+++ b/internal/testhelper/testhelper.go
@@ -128,9 +128,9 @@ func GitalyServersMetadata(t testing.TB, serverSocketPath string) metadata.MD {
 	return metadata.Pairs("gitaly-servers", base64.StdEncoding.EncodeToString(gitalyServersJSON))
 }
 
-func testRepoValid(repo *gitalypb.Repository) bool {
-	storagePath, _ := config.Config.StoragePath(repo.GetStorageName())
-	if _, err := os.Stat(path.Join(storagePath, repo.RelativePath, "objects")); err != nil {
+// isValidRepoPath checks whether a valid git repository exists at the given path.
+func isValidRepoPath(absolutePath string) bool {
+	if _, err := os.Stat(filepath.Join(absolutePath, "objects")); err != nil {
 		return false
 	}
 
@@ -148,7 +148,8 @@ func TestRepository() *gitalypb.Repository {
 		GlRepository: "project-1",
 	}
 
-	if !testRepoValid(repo) {
+	storagePath, _ := config.Config.StoragePath(repo.GetStorageName())
+	if !isValidRepoPath(filepath.Join(storagePath, repo.RelativePath)) {
 		panic("Test repo not found, did you run `make prepare-tests`?")
 	}
 
@@ -460,27 +461,39 @@ func initRepo(t testing.TB, bare bool) (*gitalypb.Repository, string, func()) {
 	return repo, repoPath, func() { require.NoError(t, os.RemoveAll(repoPath)) }
 }
 
-// NewTestRepo creates a bare copy of the test repository.
+// NewTestRepoTo clones a new copy of test repository under a subdirectory in the storage root.
+func NewTestRepoTo(t testing.TB, storageRoot, relativePath string) *gitalypb.Repository {
+	repo, _, _ := cloneTestRepo(t, storageRoot, relativePath, true)
+	return repo
+}
+
+// NewTestRepo creates a bare copy of the test repository..
 func NewTestRepo(t testing.TB) (repo *gitalypb.Repository, repoPath string, cleanup func()) {
-	return cloneTestRepo(t, true)
+	return cloneTestRepo(t, GitlabTestStoragePath(), NewRepositoryName(t, true), true)
 }
 
 // NewTestRepoWithWorktree creates a copy of the test repository with a
 // worktree. This is allows you to run normal 'non-bare' Git commands.
 func NewTestRepoWithWorktree(t testing.TB) (repo *gitalypb.Repository, repoPath string, cleanup func()) {
-	return cloneTestRepo(t, false)
+	return cloneTestRepo(t, GitlabTestStoragePath(), NewRepositoryName(t, false), false)
 }
 
-func cloneTestRepo(t testing.TB, bare bool) (repo *gitalypb.Repository, repoPath string, cleanup func()) {
-	storagePath := GitlabTestStoragePath()
-	relativePath := NewRepositoryName(t, bare)
-	repoPath = filepath.Join(storagePath, relativePath)
+// testRepositoryPath returns the absolute path of local 'gitlab-org/gitlab-test.git' clone.
+// It is cloned under the path by the test preparing step of make.
+func testRepositoryPath(t testing.TB) string {
+	path := filepath.Join(GitlabTestStoragePath(), TestRelativePath)
+	if !isValidRepoPath(path) {
+		t.Fatalf("local clone of 'gitlab-org/gitlab-test.git' not found in %q, did you run `make prepare-tests`?", path)
+	}
 
-	repo = CreateRepo(t, storagePath, relativePath)
-	testRepo := TestRepository()
-	testRepoPath := path.Join(storagePath, testRepo.RelativePath)
-	args := []string{"clone", "--no-hardlinks", "--dissociate"}
+	return path
+}
+
+func cloneTestRepo(t testing.TB, storageRoot, relativePath string, bare bool) (repo *gitalypb.Repository, repoPath string, cleanup func()) {
+	repoPath = filepath.Join(storageRoot, relativePath)
 
+	repo = CreateRepo(t, storageRoot, relativePath)
+	args := []string{"clone", "--no-hardlinks", "--dissociate"}
 	if bare {
 		args = append(args, "--bare")
 	} else {
@@ -488,7 +501,7 @@ func cloneTestRepo(t testing.TB, bare bool) (repo *gitalypb.Repository, repoPath
 		repo.RelativePath = path.Join(relativePath, ".git")
 	}
 
-	MustRunCommand(t, nil, "git", append(args, testRepoPath, repoPath)...)
+	MustRunCommand(t, nil, "git", append(args, testRepositoryPath(t), repoPath)...)
 
 	return repo, repoPath, func() { require.NoError(t, os.RemoveAll(repoPath)) }
 }