storage: Add option to skip storage check

In some cases, such as `CreateFork`, Praefect may not rewrite the
storage name to one present on the receiving Gitaly node. In these cases
we must skip validating that the storage is present on disk.

Add a `SkipStorageExistenceCheck` option to the storage locator to allow
us to skip this test when needed. This necessarily skips the repository
check as well, as we don't have a storage path to join to.

3 files changed, 86 insertions, 5 deletions

diff --git a/internal/gitaly/config/locator.go b/internal/gitaly/config/locator.go
index d11392413..f077d039f 100644
--- a/internal/gitaly/config/locator.go
+++ b/internal/gitaly/config/locator.go
@@ -53,6 +53,15 @@ func (l *configLocator) ValidateRepository(repo storage.Repository, opts ...stor
 		return structerr.NewInvalidArgument("%w", storage.ErrRepositoryNotSet)
 	}
 
+	relativePath := repo.GetRelativePath()
+	if len(relativePath) == 0 {
+		return structerr.NewInvalidArgument("%w", storage.ErrRepositoryPathNotSet)
+	}
+
+	if cfg.SkipStorageExistenceCheck {
+		return nil
+	}
+
 	storagePath, err := l.GetStorageByName(repo.GetStorageName())
 	if err != nil {
 		return err
@@ -65,11 +74,6 @@ func (l *configLocator) ValidateRepository(repo storage.Repository, opts ...stor
 		return structerr.New("storage path: %w", err).WithMetadata("storage_path", storagePath)
 	}
 
-	relativePath := repo.GetRelativePath()
-	if len(relativePath) == 0 {
-		return structerr.NewInvalidArgument("%w", storage.ErrRepositoryPathNotSet)
-	}
-
 	if _, err := storage.ValidateRelativePath(storagePath, relativePath); err != nil {
 		return structerr.NewInvalidArgument("%w", err).WithMetadata("relative_path", relativePath)
 	}
diff --git a/internal/gitaly/config/locator_test.go b/internal/gitaly/config/locator_test.go
index 17de85070..dc78fe493 100644
--- a/internal/gitaly/config/locator_test.go
+++ b/internal/gitaly/config/locator_test.go
@@ -110,6 +110,69 @@ func TestConfigLocator_GetRepoPath(t *testing.T) {
 	}
 }
 
+func TestConfigLocator_ValidateRepository(t *testing.T) {
+	t.Parallel()
+
+	ctx := testhelper.Context(t)
+
+	const storageName = "exists"
+	cfg := testcfg.Build(t, testcfg.WithStorages(storageName))
+	cfg.SocketPath = testserver.RunGitalyServer(t, cfg, setup.RegisterAll)
+	locator := config.NewLocator(cfg)
+
+	repo, repoPath := gittest.CreateRepository(t, ctx, cfg)
+	if testhelper.IsPraefectEnabled() {
+		repo.RelativePath = strings.TrimPrefix(repoPath, cfg.Storages[0].Path)
+	}
+
+	for _, tc := range []struct {
+		desc   string
+		repo   *gitalypb.Repository
+		opts   []storage.ValidateRepositoryOption
+		expErr error
+	}{
+		{
+			desc:   "unknown storage",
+			repo:   &gitalypb.Repository{StorageName: "invalid", RelativePath: repo.RelativePath},
+			expErr: structerr.NewInvalidArgument("%w", storage.ErrStorageNotFound).WithMetadata("storage_name", "invalid"),
+		},
+		{
+			desc: "unchecked unknown storage",
+			repo: &gitalypb.Repository{StorageName: "invalid", RelativePath: repo.RelativePath},
+			opts: []storage.ValidateRepositoryOption{storage.WithSkipStorageExistenceCheck()},
+		},
+		{
+			desc:   "unchecked unset storage",
+			repo:   &gitalypb.Repository{StorageName: "", RelativePath: repo.RelativePath},
+			expErr: structerr.NewInvalidArgument("%w", storage.ErrStorageNotSet),
+		},
+		{
+			desc:   "unknown repository",
+			repo:   &gitalypb.Repository{StorageName: storageName, RelativePath: "invalid"},
+			expErr: storage.NewRepositoryNotFoundError(storageName, "invalid"),
+		},
+		{
+			desc: "unchecked unknown repo",
+			repo: &gitalypb.Repository{StorageName: storageName, RelativePath: "invalid"},
+			opts: []storage.ValidateRepositoryOption{storage.WithSkipRepositoryExistenceCheck()},
+		},
+		{
+			desc:   "unchecked unset repository",
+			repo:   &gitalypb.Repository{StorageName: storageName, RelativePath: ""},
+			expErr: structerr.NewInvalidArgument("%w", storage.ErrRepositoryPathNotSet),
+		},
+		{
+			desc: "proper repository path",
+			repo: repo,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			err := locator.ValidateRepository(tc.repo, tc.opts...)
+			require.Equal(t, tc.expErr, err)
+		})
+	}
+}
+
 func TestConfigLocator_CacheDir(t *testing.T) {
 	t.Parallel()
 	const storageName = "exists"
diff --git a/internal/gitaly/storage/locator.go b/internal/gitaly/storage/locator.go
index ee63de33f..5a116de6a 100644
--- a/internal/gitaly/storage/locator.go
+++ b/internal/gitaly/storage/locator.go
@@ -95,6 +95,12 @@ type Locator interface {
 
 // ValidateRepositoryConfig is used to configure ValidateRepository.
 type ValidateRepositoryConfig struct {
+	// SkipStorageExistenceCheck determines whether the storage shall be checked for
+	// existence or not. If set, the locator shall still perform parameter verification and
+	// verify that whether the storage _would_ be valid if it existed, but not verify actual
+	// existence. This will also skip the repository existence check. Required when
+	// Praefect has not rewritten the repository storage name.
+	SkipStorageExistenceCheck bool
 	// SkipRepositoryExistenceCheck determines whether the repository shall be checked for
 	// existence or not. If set, the locator shall still perform parameter verification and
 	// verify that whether the repository _would_ be valid if it existed, but not verify actual
@@ -113,6 +119,14 @@ func WithSkipRepositoryExistenceCheck() ValidateRepositoryOption {
 	}
 }
 
+// WithSkipStorageExistenceCheck causes ValidateRepository to not check for storage
+// existence. Repository existence check will also be skipped.
+func WithSkipStorageExistenceCheck() ValidateRepositoryOption {
+	return func(cfg *ValidateRepositoryConfig) {
+		cfg.SkipStorageExistenceCheck = true
+	}
+}
+
 // GetRepoPathConfig is used to configure GetRepoPath.
 type GetRepoPathConfig struct {
 	// SkipRepositoryVerification will cause GetRepoPath to skip verification the verification whether the