Inspect changes in Gitaly 4261fb1dccde53037494d42745c59583f6644553
Upstream merge commit 9c8c31fafd35 changes Gitaly server version from 7eb79ebcb084d4e881777f44ca5055cce6e60ccf to 4261fb1dccde53037494d42745c59583f6644553, with changes in protocol or one of the source subtrees currently under watch.
To further inspect, go to an up to date Gitaly source clone and run:
git diff 7eb79ebcb084d4e881777f44ca5055cce6e60ccf..4261fb1dccde53037494d42745c59583f6644553 -- proto/*.proto internal/backup
Here is a preview, truncated at 500 lines:
diff --git a/internal/backup/backup.go b/internal/backup/backup.go
index 5eed0724a..bc54cbaf0 100644
--- a/internal/backup/backup.go
+++ b/internal/backup/backup.go
@@ -211,25 +211,56 @@ func NewManagerLocal(
}
}
-// RemoveAllRepositories removes all repositories in the specified storage name.
-func (mgr *Manager) RemoveAllRepositories(ctx context.Context, req *RemoveAllRepositoriesRequest) error {
- if err := setContextServerInfo(ctx, &req.Server, req.StorageName); err != nil {
- return fmt.Errorf("manager: %w", err)
+// RemoveRepository removes the specified repository from its storage.
+func (mgr *Manager) RemoveRepository(ctx context.Context, req *RemoveRepositoryRequest) error {
+ if err := setContextServerInfo(ctx, &req.Server, req.Repo.StorageName); err != nil {
+ return fmt.Errorf("remove repo: set context: %w", err)
}
repoClient, err := mgr.newRepoClient(ctx, req.Server)
if err != nil {
- return fmt.Errorf("manager: %w", err)
+ return fmt.Errorf("remove repo: create client: %w", err)
}
- _, err = repoClient.RemoveAll(ctx, &gitalypb.RemoveAllRequest{StorageName: req.StorageName})
+ _, err = repoClient.RemoveRepository(ctx, &gitalypb.RemoveRepositoryRequest{Repository: req.Repo})
if err != nil {
- return fmt.Errorf("manager: %w", err)
+ return fmt.Errorf("remove repo: remove: %w", err)
}
return nil
}
+// ListRepositories returns a list of repositories found in the given storage.
+func (mgr *Manager) ListRepositories(ctx context.Context, req *ListRepositoriesRequest) (repos []*gitalypb.Repository, err error) {
+ if err := setContextServerInfo(ctx, &req.Server, req.StorageName); err != nil {
+ return nil, fmt.Errorf("list repos: set context: %w", err)
+ }
+
+ internalClient, err := mgr.newInternalClient(ctx, req.Server)
+ if err != nil {
+ return nil, fmt.Errorf("list repos: create client: %w", err)
+ }
+
+ stream, err := internalClient.WalkRepos(ctx, &gitalypb.WalkReposRequest{StorageName: req.StorageName})
+ if err != nil {
+ return nil, fmt.Errorf("list repos: walk: %w", err)
+ }
+
+ for {
+ resp, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, fmt.Errorf("list repos: receiving messages: %w", err)
+ }
+
+ repos = append(repos, &gitalypb.Repository{RelativePath: resp.RelativePath, StorageName: req.StorageName})
+ }
+
+ return repos, nil
+}
+
// Create creates a repository backup.
func (mgr *Manager) Create(ctx context.Context, req *CreateRequest) error {
if req.VanityRepository == nil {
@@ -304,10 +335,6 @@ func (mgr *Manager) Restore(ctx context.Context, req *RestoreRequest) error {
return fmt.Errorf("manager: %w", err)
}
- if err := repo.Remove(ctx); err != nil {
- return fmt.Errorf("manager: %w", err)
- }
-
var backup *Backup
if req.BackupID == "" {
backup, err = mgr.locator.FindLatest(ctx, req.VanityRepository)
@@ -328,6 +355,10 @@ func (mgr *Manager) Restore(ctx context.Context, req *RestoreRequest) error {
defaultBranch, defaultBranchKnown := git.ReferenceName(backup.HeadReference).Branch()
+ if err := repo.Remove(ctx); err != nil {
+ return fmt.Errorf("manager: %w", err)
+ }
+
if err := repo.Create(ctx, hash, defaultBranch); err != nil {
return fmt.Errorf("manager: %w", err)
}
@@ -573,3 +604,12 @@ func (mgr *Manager) newRepoClient(ctx context.Context, server storage.ServerInfo
return gitalypb.NewRepositoryServiceClient(conn), nil
}
+
+func (mgr *Manager) newInternalClient(ctx context.Context, server storage.ServerInfo) (gitalypb.InternalGitalyClient, error) {
+ conn, err := mgr.conns.Dial(ctx, server.Address, server.Token)
+ if err != nil {
+ return nil, err
+ }
+
+ return gitalypb.NewInternalGitalyClient(conn), nil
+}
diff --git a/internal/backup/backup_test.go b/internal/backup/backup_test.go
index 14341ec6a..96a06fba3 100644
--- a/internal/backup/backup_test.go
+++ b/internal/backup/backup_test.go
@@ -28,13 +28,10 @@ import (
"google.golang.org/protobuf/proto"
)
-func TestManager_RemoveAllRepositories(t *testing.T) {
- testhelper.SkipWithWAL(t, `
-RemoveAll is removing the entire content of the storage. This would also remove the database's and
-the transaction manager's disk state. The RPC needs to be updated to shut down all partitions and
-the database and only then perform the removal.
-
-Issue: https://gitlab.com/gitlab-org/gitaly/-/issues/5269`)
+func TestManager_RemoveRepository(t *testing.T) {
+ if testhelper.IsPraefectEnabled() {
+ t.Skip("local backup manager expects to operate on the local filesystem so cannot operate through praefect")
+ }
t.Parallel()
@@ -58,11 +55,109 @@ Issue: https://gitlab.com/gitlab-org/gitaly/-/issues/5269`)
require.NoError(t, err)
fsBackup := backup.NewManager(sink, locator, pool)
- err = fsBackup.RemoveAllRepositories(ctx, &backup.RemoveAllRepositoriesRequest{
- Server: storage.ServerInfo{Address: cfg.SocketPath, Token: cfg.Auth.Token},
- StorageName: repo.StorageName,
+ err = fsBackup.RemoveRepository(ctx, &backup.RemoveRepositoryRequest{
+ Server: storage.ServerInfo{Address: cfg.SocketPath, Token: cfg.Auth.Token},
+ Repo: repo,
})
require.NoError(t, err)
+ require.NoDirExists(t, repoPath)
+
+ // With an invalid repository
+ err = fsBackup.RemoveRepository(ctx, &backup.RemoveRepositoryRequest{
+ Server: storage.ServerInfo{Address: cfg.SocketPath, Token: cfg.Auth.Token},
+ Repo: &gitalypb.Repository{StorageName: "nonexistent", RelativePath: "nonexistent"},
+ })
+
+ require.EqualError(t, err, "remove repo: remove: rpc error: code = InvalidArgument desc = storage name not found")
+}
+
+func TestManager_ListRepositories(t *testing.T) {
+ t.Parallel()
+
+ for _, tc := range []struct {
+ desc string
+ repos map[string][]*gitalypb.Repository
+ }{
+ {
+ desc: "no repos",
+ repos: make(map[string][]*gitalypb.Repository),
+ },
+ {
+ desc: "repos in a single storage",
+ repos: map[string][]*gitalypb.Repository{
+ "storage-1": {
+ {RelativePath: "a", StorageName: "storage-1"},
+ {RelativePath: "b", StorageName: "storage-1"},
+ },
+ },
+ },
+ {
+ desc: "repos in multiple storages",
+ repos: map[string][]*gitalypb.Repository{
+ "storage-1": {
+ {RelativePath: "a", StorageName: "storage-1"},
+ {RelativePath: "b", StorageName: "storage-1"},
+ },
+ "storage-2": {
+ {RelativePath: "c", StorageName: "storage-2"},
+ {RelativePath: "d", StorageName: "storage-2"},
+ },
+ },
+ },
+ } {
+ t.Run(tc.desc, func(t *testing.T) {
+ if testhelper.IsPraefectEnabled() {
+ t.Skip("local backup manager expects to operate on the local filesystem so cannot operate through praefect")
+ }
+
+ var storages []string
+ for storageName := range tc.repos {
+ storages = append(storages, storageName)
+ }
+
+ // We don't really need a "default" storage, but this makes initialisation cleaner since
+ // WithStorages() takes at least one argument.
+ cfg := testcfg.Build(t, testcfg.WithStorages("default", storages...))
+ cfg.SocketPath = testserver.RunGitalyServer(t, cfg, setup.RegisterAll)
+
+ ctx := testhelper.Context(t)
+
+ for storageName, repos := range tc.repos {
+ for _, repo := range repos {
+ storagePath, ok := cfg.StoragePath(storageName)
+ require.True(t, ok)
+
+ _, _ = gittest.CreateRepository(t, ctx, cfg, gittest.CreateRepositoryConfig{
+ SkipCreationViaService: true,
+ RelativePath: repo.RelativePath,
+ Storage: config.Storage{Name: storageName, Path: storagePath},
+ })
+ }
+ }
+
+ pool := client.NewPool()
+ defer testhelper.MustClose(t, pool)
+
+ backupRoot := testhelper.TempDir(t)
+ sink := backup.NewFilesystemSink(backupRoot)
+ defer testhelper.MustClose(t, sink)
+
+ locator, err := backup.ResolveLocator("pointer", sink)
+ require.NoError(t, err)
+
+ fsBackup := backup.NewManager(sink, locator, pool)
+
+ for storageName, repos := range tc.repos {
+ actualRepos, err := fsBackup.ListRepositories(ctx, &backup.ListRepositoriesRequest{
+ Server: storage.ServerInfo{Address: cfg.SocketPath, Token: cfg.Auth.Token},
+ StorageName: storageName,
+ })
+
+ require.NoError(t, err)
+ require.EqualValues(t, repos, actualRepos)
+ }
+ })
+ }
}
func TestManager_Create(t *testing.T) {
@@ -882,6 +977,7 @@ func TestManager_Restore_specific(t *testing.T) {
return repo, nil
},
expectedErrAs: backup.ErrSkipped,
+ expectExists: true,
},
{
desc: "single incremental",
diff --git a/internal/backup/locator.go b/internal/backup/locator.go
index 1e1c42550..c0ccca3b3 100644
--- a/internal/backup/locator.go
+++ b/internal/backup/locator.go
@@ -267,6 +267,8 @@ func (l PointerLocator) writeLatest(ctx context.Context, path, target string) (r
return nil
}
+const latestManifestName = "+latest"
+
// ManifestLocator locates backup paths based on manifest files that are
// written to a predetermined path:
//
@@ -294,44 +296,55 @@ func (l ManifestLocator) Commit(ctx context.Context, backup *Backup) (returnErr
return err
}
- f, err := l.Sink.GetWriter(ctx, manifestPath(backup.Repository, backup.ID))
- if err != nil {
+ if err := l.writeManifest(ctx, backup, backup.ID); err != nil {
return fmt.Errorf("manifest: commit: %w", err)
}
- defer func() {
- if err := f.Close(); err != nil && returnErr == nil {
- returnErr = fmt.Errorf("manifest: commit: %w", err)
- }
- }()
-
- if err := toml.NewEncoder(f).Encode(backup); err != nil {
- return fmt.Errorf("manifest: commit: %w", err)
+ if err := l.writeManifest(ctx, backup, latestManifestName); err != nil {
+ return fmt.Errorf("manifest: commit latest: %w", err)
}
return nil
}
-// FindLatest passes through to Fallback
+// FindLatest loads the manifest called +latest. If this manifest does not
+// exist, the Fallback is used.
func (l ManifestLocator) FindLatest(ctx context.Context, repo storage.Repository) (*Backup, error) {
- return l.Fallback.FindLatest(ctx, repo)
+ backup, err := l.readManifest(ctx, repo, latestManifestName)
+ switch {
+ case errors.Is(err, ErrDoesntExist):
+ return l.Fallback.FindLatest(ctx, repo)
+ case err != nil:
+ return nil, fmt.Errorf("manifest: find latest: %w", err)
+ }
+
+ return backup, nil
}
// Find loads the manifest for the provided repo and backupID. If this manifest
// does not exist, the fallback is used.
func (l ManifestLocator) Find(ctx context.Context, repo storage.Repository, backupID string) (*Backup, error) {
- f, err := l.Sink.GetReader(ctx, manifestPath(repo, backupID))
+ backup, err := l.readManifest(ctx, repo, backupID)
switch {
case errors.Is(err, ErrDoesntExist):
return l.Fallback.Find(ctx, repo, backupID)
case err != nil:
return nil, fmt.Errorf("manifest: find: %w", err)
}
+
+ return backup, nil
+}
+
+func (l ManifestLocator) readManifest(ctx context.Context, repo storage.Repository, backupID string) (*Backup, error) {
+ f, err := l.Sink.GetReader(ctx, manifestPath(repo, backupID))
+ if err != nil {
+ return nil, fmt.Errorf("read manifest: %w", err)
+ }
defer f.Close()
var backup Backup
if err := toml.NewDecoder(f).Decode(&backup); err != nil {
- return nil, fmt.Errorf("manifest: find: %w", err)
+ return nil, fmt.Errorf("read manifest: %w", err)
}
backup.ID = backupID
@@ -340,6 +353,24 @@ func (l ManifestLocator) Find(ctx context.Context, repo storage.Repository, back
return &backup, nil
}
+func (l ManifestLocator) writeManifest(ctx context.Context, backup *Backup, backupID string) (returnErr error) {
+ f, err := l.Sink.GetWriter(ctx, manifestPath(backup.Repository, backupID))
+ if err != nil {
+ return fmt.Errorf("write manifest: %w", err)
+ }
+ defer func() {
+ if err := f.Close(); err != nil && returnErr == nil {
+ returnErr = fmt.Errorf("write manifest: %w", err)
+ }
+ }()
+
+ if err := toml.NewEncoder(f).Encode(backup); err != nil {
+ return fmt.Errorf("write manifest: %w", err)
+ }
+
+ return nil
+}
+
func manifestPath(repo storage.Repository, backupID string) string {
storageName := repo.GetStorageName()
// Other locators strip the .git suffix off of relative paths. This suffix
diff --git a/internal/backup/locator_test.go b/internal/backup/locator_test.go
index 90cac3664..1d6010554 100644
--- a/internal/backup/locator_test.go
+++ b/internal/backup/locator_test.go
@@ -470,7 +470,9 @@ custom_hooks_path = '%[1]s/%[2]s/001.custom_hooks.tar'
require.NoError(t, l.Commit(ctx, incremental))
manifest := testhelper.MustReadFile(t, filepath.Join(backupPath, "manifests", repo.StorageName, repo.RelativePath, backupID+".toml"))
- require.Equal(t, fmt.Sprintf(`object_format = 'sha1'
+ latestManifest := testhelper.MustReadFile(t, filepath.Join(backupPath, "manifests", repo.StorageName, repo.RelativePath, "+latest.toml"))
+
+ expectedManifest := fmt.Sprintf(`object_format = 'sha1'
[[steps]]
bundle_path = '%[1]s/%[2]s/001.bundle'
@@ -482,7 +484,10 @@ bundle_path = '%[1]s/%[2]s/002.bundle'
ref_path = '%[1]s/%[2]s/002.refs'
previous_ref_path = '%[1]s/%[2]s/001.refs'
custom_hooks_path = '%[1]s/%[2]s/002.custom_hooks.tar'
-`, repo.RelativePath, backupID), string(manifest))
+`, repo.RelativePath, backupID)
+
+ require.Equal(t, expectedManifest, string(manifest))
+ require.Equal(t, expectedManifest, string(latestManifest))
})
}
@@ -605,3 +610,120 @@ custom_hooks_path = 'path/to/002.custom_hooks.tar'
})
}
}
+
+func TestManifestLocator_FindLatest(t *testing.T) {
+ t.Parallel()
+
+ for _, tc := range []struct {
+ desc string
+ repo storage.Repository
+ setup func(t *testing.T, ctx context.Context, backupPath string)
+ expectedBackup *Backup
+ }{
+ {
+ desc: "finds manifest",
+ repo: &gitalypb.Repository{
+ StorageName: "default",
+ RelativePath: "vanity/repo.git",
+ },
+ setup: func(t *testing.T, ctx context.Context, backupPath string) {
+ testhelper.WriteFiles(t, backupPath, map[string]any{
+ "vanity/repo/LATEST": "abc123",
+ "vanity/repo/abc123/LATEST": "002",
+ "manifests/default/vanity/repo.git/+latest.toml": `object_format = 'sha1'
+
+[[steps]]
+bundle_path = 'manifest-path/to/001.bundle'
+ref_path = 'manifest-path/to/001.refs'
+custom_hooks_path = 'manifest-path/to/001.custom_hooks.tar'
+
+[[steps]]
+bundle_path = 'manifest-path/to/002.bundle'
+ref_path = 'manifest-path/to/002.refs'
+previous_ref_path = 'manifest-path/to/001.refs'
+custom_hooks_path = 'manifest-path/to/002.custom_hooks.tar'
+`,
+ })
+ },
+ expectedBackup: &Backup{
+ ID: "+latest",
+ Repository: &gitalypb.Repository{
+ StorageName: "default",
+ RelativePath: "vanity/repo.git",
+ },
+ ObjectFormat: "sha1",
+ Steps: []Step{
+ {
+ BundlePath: "manifest-path/to/001.bundle",
+ RefPath: "manifest-path/to/001.refs",
+ CustomHooksPath: "manifest-path/to/001.custom_hooks.tar",
+ },
+ {
+ BundlePath: "manifest-path/to/002.bundle",
+ RefPath: "manifest-path/to/002.refs",
+ PreviousRefPath: "manifest-path/to/001.refs",
+ CustomHooksPath: "manifest-path/to/002.custom_hooks.tar",
+ },
+ },
+ },
+ },
+ {
+ desc: "fallback",
+ repo: &gitalypb.Repository{
+ StorageName: "default",
+ RelativePath: "vanity/repo.git",
+ },
+ setup: func(t *testing.T, ctx context.Context, backupPath string) {
+ testhelper.WriteFiles(t, backupPath, map[string]any{
+ "vanity/repo/LATEST": "abc123",
+ "vanity/repo/abc123/LATEST": "002",
+ })
+ },
+ expectedBackup: &Backup{
+ ID: "abc123",
+ Repository: &gitalypb.Repository{
+ StorageName: "default",
+ RelativePath: "vanity/repo.git",
+ },
+ ObjectFormat: "sha1",
+ Steps: []Step{
+ {
+ BundlePath: "vanity/repo/abc123/001.bundle",
+ RefPath: "vanity/repo/abc123/001.refs",
+ CustomHooksPath: "vanity/repo/abc123/001.custom_hooks.tar",
+ },
+ {
+ BundlePath: "vanity/repo/abc123/002.bundle",
+ RefPath: "vanity/repo/abc123/002.refs",
+ PreviousRefPath: "vanity/repo/abc123/001.refs",
+ CustomHooksPath: "vanity/repo/abc123/002.custom_hooks.tar",
+ },
+ },
+ },
+ },
+ } {
+ tc := tc
+ t.Run(tc.desc, func(t *testing.T) {
+ t.Parallel()
+
+ ctx := testhelper.Context(t)
+ backupPath := testhelper.TempDir(t)
+
+ tc.setup(t, ctx, backupPath)
+
+ sink := NewFilesystemSink(backupPath)
+ var l Locator = PointerLocator{
+ Sink: sink,
+ }
+ l = ManifestLocator{
+ Sink: sink,
+ Fallback: l,
+ }
+
+ backup, err := l.FindLatest(ctx, tc.repo)
+ require.NoError(t, err)
+
+ require.Equal(t, tc.expectedBackup, backup)
+ })
A merge request may have already been submitted, check that on the merge commit, as usually changes in Gitaly are not breaking, compliance becoming mandatory in the following development cycles.