Inspect changes in Gitaly 720261882a2bbd549cdbccded8ea9b3ca5af846f
Upstream merge commit f47cdf09de1c changes Gitaly server version from 84074347b5ad5ff09fd4d6068c8bcdf356f7798c to 720261882a2bbd549cdbccded8ea9b3ca5af846f, with changes in protocol or one of the source subtrees currently under watch.
The topic upstream-merge
is at 63 non-merge commits.
To further inspect, go to an up to date Gitaly source clone and run:
git diff 84074347b5ad5ff09fd4d6068c8bcdf356f7798c..720261882a2bbd549cdbccded8ea9b3ca5af846f -- proto/*.proto internal/backup
Here is a preview, truncated at 500 lines:
diff --git a/internal/backup/backup_test.go b/internal/backup/backup_test.go
index 2b30180ca..922fa395b 100644
--- a/internal/backup/backup_test.go
+++ b/internal/backup/backup_test.go
@@ -205,7 +205,8 @@ func TestManager_Create(t *testing.T) {
bundlePath := joinBackupPath(t, backupRoot, vanityRepo, backupID, "001.bundle")
customHooksPath := joinBackupPath(t, backupRoot, vanityRepo, backupID, "001.custom_hooks.tar")
- sink := backup.NewFilesystemSink(backupRoot)
+ sink, err := backup.ResolveSink(ctx, backupRoot)
+ require.NoError(t, err)
defer testhelper.MustClose(t, sink)
locator, err := backup.ResolveLocator("manifest", sink)
@@ -230,11 +231,11 @@ func TestManager_Create(t *testing.T) {
dirInfo, err := os.Stat(filepath.Dir(bundlePath))
require.NoError(t, err)
- require.Equal(t, perm.PrivateDir, dirInfo.Mode().Perm(), "expecting restricted directory permissions")
+ require.Equal(t, perm.SharedDir, dirInfo.Mode().Perm(), "expecting shared directory permissions")
bundleInfo, err := os.Stat(bundlePath)
require.NoError(t, err)
- require.Equal(t, perm.PrivateFile, bundleInfo.Mode().Perm(), "expecting restricted file permissions")
+ require.Equal(t, perm.SharedFile, bundleInfo.Mode().Perm(), "expecting shared file permissions")
output := gittest.Exec(t, cfg, "-C", data.repoPath, "bundle", "verify", bundlePath)
require.Contains(t, string(output), "The bundle records a complete history")
@@ -391,7 +392,8 @@ custom_hooks_path = '%[2]s/001.custom_hooks.tar'
refsPath := joinBackupPath(t, backupRoot, repo, backupID, tc.expectedIncrement+".refs")
bundlePath := joinBackupPath(t, backupRoot, repo, backupID, tc.expectedIncrement+".bundle")
- sink := backup.NewFilesystemSink(backupRoot)
+ sink, err := backup.ResolveSink(ctx, backupRoot)
+ require.NoError(t, err)
defer testhelper.MustClose(t, sink)
locator, err := backup.ResolveLocator("manifest", sink)
@@ -762,7 +764,8 @@ custom_hooks_path = '%[2]s/%[3]s/002.custom_hooks.tar'
t.Run(tc.desc, func(t *testing.T) {
repo, expectedChecksum := tc.setup(t)
- sink := backup.NewFilesystemSink(backupRoot)
+ sink, err := backup.ResolveSink(ctx, backupRoot)
+ require.NoError(t, err)
defer testhelper.MustClose(t, sink)
locator, err := backup.ResolveLocator("manifest", sink)
@@ -1061,7 +1064,8 @@ custom_hooks_path = 'custom_hooks.tar'
t.Run(tc.desc, func(t *testing.T) {
repo, expectedChecksum := tc.setup(t)
- sink := backup.NewFilesystemSink(backupRoot)
+ sink, err := backup.ResolveSink(ctx, backupRoot)
+ require.NoError(t, err)
defer testhelper.MustClose(t, sink)
locator, err := backup.ResolveLocator("manifest", sink)
@@ -1139,7 +1143,8 @@ func TestManager_CreateRestore_contextServerInfo(t *testing.T) {
pool := client.NewPool()
defer testhelper.MustClose(t, pool)
- sink := backup.NewFilesystemSink(backupRoot)
+ sink, err := backup.ResolveSink(ctx, backupRoot)
+ require.NoError(t, err)
defer testhelper.MustClose(t, sink)
locator, err := backup.ResolveLocator("manifest", sink)
diff --git a/internal/backup/filesystem_sink.go b/internal/backup/filesystem_sink.go
deleted file mode 100644
index 437571fd2..000000000
--- a/internal/backup/filesystem_sink.go
+++ /dev/null
@@ -1,84 +0,0 @@
-package backup
-
-import (
- "context"
- "errors"
- "fmt"
- "io"
- "os"
- "path/filepath"
- "time"
-
- "gitlab.com/gitlab-org/gitaly/v16/internal/helper/perm"
- "gitlab.com/gitlab-org/gitaly/v16/internal/safe"
- "gitlab.com/gitlab-org/gitaly/v16/internal/structerr"
-)
-
-// FilesystemSink is a sink for creating and restoring backups from the local filesystem.
-type FilesystemSink struct {
- path string
-}
-
-// NewFilesystemSink returns a sink that uses a local filesystem to work with data.
-func NewFilesystemSink(path string) *FilesystemSink {
- return &FilesystemSink{
- path: path,
- }
-}
-
-// GetWriter opens a io.WriteCloser that can be used to write data into a
-// relativePath path on the filesystem. It is the callers responsibility to
-// Close the writer after usage.
-func (fs *FilesystemSink) GetWriter(ctx context.Context, relativePath string) (io.WriteCloser, error) {
- path := filepath.Join(fs.path, relativePath)
-
- if err := os.MkdirAll(filepath.Dir(path), perm.PrivateDir); err != nil {
- return nil, fmt.Errorf("filesystem sink: %w", err)
- }
-
- f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, perm.PrivateFile)
- if err != nil {
- return nil, fmt.Errorf("filesystem sink: %w", err)
- }
- return f, nil
-}
-
-// GetReader returns a reader of the requested file path.
-// It's the caller's responsibility to Close returned reader once it is not needed anymore.
-// If relativePath doesn't exist the ErrDoesntExist is returned.
-func (fs *FilesystemSink) GetReader(ctx context.Context, relativePath string) (io.ReadCloser, error) {
- path := filepath.Join(fs.path, relativePath)
- f, err := os.Open(path)
- if err != nil {
- if errors.Is(err, os.ErrNotExist) {
- return nil, ErrDoesntExist
- }
- return nil, fmt.Errorf("filesystem sink: %w", err)
- }
- return f, nil
-}
-
-// Close will fsync the written file and its parent directory.
-func (fs *FilesystemSink) Close() error {
- _, err := os.Stat(fs.path)
- if err != nil {
- if errors.Is(err, os.ErrNotExist) {
- return nil
- }
- return fmt.Errorf("stat sink path: %w", err)
- }
-
- syncer := safe.NewSyncer()
- if err := syncer.Sync(fs.path); err != nil {
- return fmt.Errorf("sync sink: %w", err)
- }
- if err := syncer.SyncParent(fs.path); err != nil {
- return fmt.Errorf("sync sink parent: %w", err)
- }
- return nil
-}
-
-// SignedURL is not supported by FilesystemSink.
-func (fs *FilesystemSink) SignedURL(ctx context.Context, relativePath string, expiry time.Duration) (string, error) {
- return "", structerr.NewUnimplemented("SignedURL not implemented for FilesystemSink")
-}
diff --git a/internal/backup/filesystem_sink_test.go b/internal/backup/filesystem_sink_test.go
deleted file mode 100644
index b06edcfdf..000000000
--- a/internal/backup/filesystem_sink_test.go
+++ /dev/null
@@ -1,115 +0,0 @@
-package backup
-
-import (
- "fmt"
- "io"
- "os"
- "path/filepath"
- "strings"
- "testing"
-
- "github.com/stretchr/testify/require"
- "gitlab.com/gitlab-org/gitaly/v16/internal/helper/perm"
- "gitlab.com/gitlab-org/gitaly/v16/internal/testhelper"
-)
-
-func TestFilesystemSink_GetReader(t *testing.T) {
- t.Parallel()
-
- t.Run("ok", func(t *testing.T) {
- t.Parallel()
- ctx := testhelper.Context(t)
-
- dir := testhelper.TempDir(t)
- const relativePath = "test.dat"
- require.NoError(t, os.WriteFile(filepath.Join(dir, relativePath), []byte("test"), perm.SharedFile))
-
- fsSink := NewFilesystemSink(dir)
- reader, err := fsSink.GetReader(ctx, relativePath)
- require.NoError(t, err)
-
- defer func() { require.NoError(t, reader.Close()) }()
-
- data, err := io.ReadAll(reader)
- require.NoError(t, err)
- require.Equal(t, []byte("test"), data)
- })
-
- t.Run("no file", func(t *testing.T) {
- t.Parallel()
- ctx := testhelper.Context(t)
-
- dir, err := os.Getwd()
- require.NoError(t, err)
-
- fsSink := NewFilesystemSink(dir)
- reader, err := fsSink.GetReader(ctx, "not-existing")
- require.Equal(t, ErrDoesntExist, err)
- require.Nil(t, reader)
- })
-}
-
-func TestFilesystemSink_GetWriter(t *testing.T) {
- t.Parallel()
-
- t.Run("ok", func(t *testing.T) {
- t.Parallel()
- ctx := testhelper.Context(t)
-
- dir := testhelper.TempDir(t)
- const relativePath = "nested/dir/test.dat"
-
- fsSink := NewFilesystemSink(dir)
- w, err := fsSink.GetWriter(ctx, relativePath)
- require.NoError(t, err)
-
- _, err = io.Copy(w, strings.NewReader("test"))
- require.NoError(t, err)
-
- require.NoError(t, w.Close())
-
- require.FileExists(t, filepath.Join(dir, relativePath))
- data, err := os.ReadFile(filepath.Join(dir, relativePath))
- require.NoError(t, err)
- require.Equal(t, []byte("test"), data)
- })
-
- t.Run("overrides existing data", func(t *testing.T) {
- t.Parallel()
- ctx := testhelper.Context(t)
-
- dir := testhelper.TempDir(t)
- const relativePath = "nested/dir/test.dat"
- fullPath := filepath.Join(dir, relativePath)
-
- require.NoError(t, os.MkdirAll(filepath.Dir(fullPath), perm.SharedDir))
- require.NoError(t, os.WriteFile(fullPath, []byte("initial"), perm.SharedFile))
-
- fsSink := NewFilesystemSink(dir)
- w, err := fsSink.GetWriter(ctx, relativePath)
- require.NoError(t, err)
-
- _, err = io.Copy(w, strings.NewReader("test"))
- require.NoError(t, err)
-
- require.NoError(t, w.Close())
-
- require.FileExists(t, fullPath)
- data, err := os.ReadFile(fullPath)
- require.NoError(t, err)
- require.Equal(t, []byte("test"), data)
- })
-
- t.Run("dir creation error", func(t *testing.T) {
- t.Parallel()
- ctx := testhelper.Context(t)
-
- dir := testhelper.TempDir(t)
- const relativePath = "nested/test.dat"
- require.NoError(t, os.WriteFile(filepath.Join(dir, "nested"), []byte("lock"), perm.PublicFile))
-
- fsSink := NewFilesystemSink(dir)
- _, err := fsSink.GetWriter(ctx, relativePath)
- require.EqualError(t, err, fmt.Sprintf(`filesystem sink: mkdir %s: not a directory`, filepath.Join(dir, "nested")))
- })
-}
diff --git a/internal/backup/locator_test.go b/internal/backup/locator_test.go
index fd633f01a..adbef4280 100644
--- a/internal/backup/locator_test.go
+++ b/internal/backup/locator_test.go
@@ -95,8 +95,10 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
const expectedIncrement = "001"
@@ -157,7 +159,8 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{Sink: sink}
if tc.setup != nil {
@@ -207,11 +210,13 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
- _, err := l.FindLatest(ctx, repo)
+ _, err = l.FindLatest(ctx, repo)
require.ErrorIs(t, err, ErrDoesntExist)
require.NoError(t, os.MkdirAll(filepath.Join(backupPath, repo.RelativePath, backupID), perm.SharedDir))
@@ -251,8 +256,10 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
Fallback: LegacyLocator{},
}
@@ -298,28 +305,32 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
- _, err := l.FindLatest(ctx, repo)
+ _, err = l.FindLatest(ctx, repo)
require.ErrorIs(t, err, ErrDoesntExist)
require.NoError(t, os.MkdirAll(filepath.Join(backupPath, repo.RelativePath), perm.SharedDir))
require.NoError(t, os.WriteFile(filepath.Join(backupPath, repo.RelativePath, "LATEST"), []byte("invalid"), perm.SharedFile))
_, err = l.FindLatest(ctx, repo)
- require.EqualError(t, err, "pointer locator: find latest: find: find latest ID: doesn't exist")
+ require.EqualError(t, err, "pointer locator: find latest: find: find latest ID: storage service sink: new reader for \"TestPointerLocator/invalid/LATEST\": doesn't exist")
})
t.Run("invalid incremental LATEST", func(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
- _, err := l.FindLatest(ctx, repo)
+ _, err = l.FindLatest(ctx, repo)
require.ErrorIs(t, err, ErrDoesntExist)
require.NoError(t, os.MkdirAll(filepath.Join(backupPath, repo.RelativePath, backupID), perm.SharedDir))
@@ -338,11 +349,13 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
- _, err := l.Find(ctx, repo, backupID)
+ _, err = l.Find(ctx, repo, backupID)
require.ErrorIs(t, err, ErrDoesntExist)
})
@@ -350,8 +363,10 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
require.NoError(t, os.MkdirAll(filepath.Join(backupPath, repo.RelativePath, backupID), perm.SharedDir))
@@ -390,11 +405,13 @@ func TestPointerLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
- Sink: NewFilesystemSink(backupPath),
+ Sink: sink,
}
- _, err := l.Find(ctx, repo, backupID)
+ _, err = l.Find(ctx, repo, backupID)
require.ErrorIs(t, err, ErrDoesntExist)
require.NoError(t, os.MkdirAll(filepath.Join(backupPath, repo.RelativePath, backupID), perm.SharedDir))
@@ -424,7 +441,8 @@ func TestManifestLocator_withFallback(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
Sink: sink,
}
@@ -456,7 +474,8 @@ custom_hooks_path = '%[1]s/%[2]s/001.custom_hooks.tar'
filepath.Join(repo.RelativePath, "abc123", "LATEST"): "001",
})
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
Sink: sink,
}
@@ -509,7 +528,8 @@ func TestManifestLocator(t *testing.T) {
t.Parallel()
backupPath := testhelper.TempDir(t)
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = ManifestLocator{
Sink: sink,
}
@@ -543,7 +563,8 @@ custom_hooks_path = '%[1]s/%[2]s/%[3]s/001.custom_hooks.tar'
`, repo.StorageName, repo.RelativePath, backupID),
})
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = ManifestLocator{
Sink: sink,
}
@@ -677,7 +698,8 @@ custom_hooks_path = 'path/to/002.custom_hooks.tar'
tc.setup(t, ctx, backupPath)
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
Sink: sink,
}
@@ -794,7 +816,8 @@ custom_hooks_path = 'manifest-path/to/002.custom_hooks.tar'
tc.setup(t, ctx, backupPath)
- sink := NewFilesystemSink(backupPath)
+ sink, err := ResolveSink(ctx, backupPath)
+ require.NoError(t, err)
var l Locator = PointerLocator{
Sink: sink,
}
diff --git a/internal/backup/log_entry.go b/internal/backup/log_entry.go
index d54538533..c0dc94b26 100644
--- a/internal/backup/log_entry.go
+++ b/internal/backup/log_entry.go
@@ -5,7 +5,6 @@ import (
"context"
"errors"
"fmt"
- "os"
"path/filepath"
"sync"
"time"
@@ -426,41 +425,19 @@ func (la *LogEntryArchiver) backupLogEntry(ctx context.Context, archiveRelPath s
// checkForExistingBackup checks if a non-zero sized file or blob exists at the archive path
// of the log entry.
func (la *LogEntryArchiver) checkForExistingBackup(ctx context.Context, archiveRelPath string) (exists bool, returnErr error) {
- switch sink := la.archiveSink.(type) {
- case *FilesystemSink:
- stat, err := os.Stat(filepath.Join(sink.path, archiveRelPath))
- if err != nil {
- if errors.Is(err, os.ErrNotExist) {
- return false, nil
- }
- return false, fmt.Errorf("stat log entry backup: %w", err)
- }
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.