From 8b86223019c8449b5adbb55e3511e1a242398faa Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Mon, 1 Jun 2026 19:25:00 +0000 Subject: [PATCH 1/5] feat: add manifest.json to eliminate S3 Walk during restore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit During upload, record every file (path, size, last-modified) in a manifest.json stored alongside metadata.json. During restore, download the manifest first and use it to enumerate files directly via GetObject, skipping the expensive recursive Walk (ListObjectsV2). For a backup with 50k files across 500 parts, this eliminates ~50 ListObjectsV2 pages of 1000 keys each, replacing them with a single GetObject for the ~2MB manifest. Key design decisions: - Manifest is built incrementally during upload (thread-safe via mutex) - Graceful fallback: older backups without manifest.json transparently fall back to Walk - Manifest upload failure is non-fatal (logged as warning) - ManifestEntry implements RemoteFile interface for seamless integration with existing download code paths - No changes to compressed (tar.gz) backup format — manifest optimization applies to DirectoryFormat where Walk is the bottleneck Closes #1375 --- pkg/backup/backuper.go | 36 +++++ pkg/backup/download.go | 26 ++- pkg/backup/upload.go | 17 ++ pkg/storage/manifest.go | 243 ++++++++++++++++++++++++++++ pkg/storage/manifest_test.go | 304 +++++++++++++++++++++++++++++++++++ 5 files changed, 624 insertions(+), 2 deletions(-) create mode 100644 pkg/storage/manifest.go create mode 100644 pkg/storage/manifest_test.go diff --git a/pkg/backup/backuper.go b/pkg/backup/backuper.go index f78bad77..cae6667c 100644 --- a/pkg/backup/backuper.go +++ b/pkg/backup/backuper.go @@ -11,6 +11,7 @@ import ( "regexp" "strings" "sync" + "time" "github.com/Altinity/clickhouse-backup/v2/pkg/common" "github.com/Altinity/clickhouse-backup/v2/pkg/metadata" @@ -51,6 +52,8 @@ type Backuper struct { resumableState *resumable.State shadowBackupUUIDs []string shadowBackupUUIDsMutex sync.Mutex + fileManifest *storage.BackupManifest + fileManifestMu sync.Mutex } func NewBackuper(cfg *config.Config, opts ...BackuperOpt) *Backuper { @@ -569,6 +572,39 @@ func (b *Backuper) addShadowBackupUUID(uuid string) { b.shadowBackupUUIDsMutex.Unlock() } +// recordUploadedFile records a single file in the backup manifest (thread-safe). +// remotePath should be the full remote path including the backupName prefix. +func (b *Backuper) recordUploadedFile(backupName, remotePath string, size int64) { + if b.fileManifest == nil { + return + } + relPath := strings.TrimPrefix(remotePath, backupName+"/") + b.fileManifestMu.Lock() + b.fileManifest.AddFile(relPath, size, time.Now().UTC()) + b.fileManifestMu.Unlock() +} + +// recordUploadedLocalFiles records multiple files in the backup manifest by +// stating the local source files to obtain their sizes (thread-safe). +// remotePath is the remote directory (including backupName prefix), localBasePath +// is the local directory the files are relative to. +func (b *Backuper) recordUploadedLocalFiles(backupName, remotePath, localBasePath string, files []string) { + if b.fileManifest == nil { + return + } + b.fileManifestMu.Lock() + defer b.fileManifestMu.Unlock() + for _, f := range files { + fullLocal := path.Join(localBasePath, f) + info, err := os.Stat(fullLocal) + if err != nil { + continue + } + relPath := strings.TrimPrefix(path.Join(remotePath, f), backupName+"/") + b.fileManifest.AddFile(relPath, info.Size(), info.ModTime()) + } +} + // CheckDisksUsage - https://github.com/Altinity/clickhouse-backup/issues/878 func (b *Backuper) CheckDisksUsage(backup storage.Backup, disks []clickhouse.Disk, isResumeExists bool, tablePattern string) error { if tablePattern != "" && tablePattern != "*.*" && tablePattern != "*" { diff --git a/pkg/backup/download.go b/pkg/backup/download.go index 28a8a8ee..1524033d 100644 --- a/pkg/backup/download.go +++ b/pkg/backup/download.go @@ -165,6 +165,10 @@ func (b *Backuper) Download(backupName string, tablePattern string, partitions [ if !found { return errors.Errorf("'%s' is not found on remote storage", backupName) } + // Download file manifest for Walk-free restore (falls back gracefully if not present) + var backupManifest *storage.BackupManifest + backupManifest, _ = b.dst.DownloadManifest(ctx, backupName) + if len(remoteBackup.Tables) == 0 && remoteBackup.RBACSize == 0 && remoteBackup.ConfigSize == 0 && remoteBackup.NamedCollectionsSize == 0 && !b.cfg.General.AllowEmptyBackups { return errors.Errorf("'%s' is empty backup", backupName) } @@ -266,7 +270,7 @@ func (b *Backuper) Download(backupName string, tablePattern string, partitions [ }).Msg("download table start") var downloadDataErr error var downloadDataSize uint64 - downloadDataSize, downloadDataErr = b.downloadTableData(dataCtx, remoteBackup.BackupMetadata, *tableMetadataAfterDownload[idx], disks, hardlinkExistsFiles) + downloadDataSize, downloadDataErr = b.downloadTableData(dataCtx, remoteBackup.BackupMetadata, *tableMetadataAfterDownload[idx], disks, hardlinkExistsFiles, backupManifest) if downloadDataErr != nil { return errors.WithMessage(downloadDataErr, "downloadTableData") } @@ -720,7 +724,7 @@ func (b *Backuper) downloadBackupRelatedDir(ctx context.Context, remoteBackup st return uint64(remoteFileInfo.Size()), nil } -func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.BackupMetadata, table metadata.TableMetadata, disks []clickhouse.Disk, hardlinkExistsFiles bool) (uint64, error) { +func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.BackupMetadata, table metadata.TableMetadata, disks []clickhouse.Disk, hardlinkExistsFiles bool, manifest *storage.BackupManifest) (uint64, error) { dbAndTableDir := path.Join(common.TablePathEncode(table.Database), common.TablePathEncode(table.Table)) dataGroup, dataCtx := errgroup.WithContext(ctx) dataGroup.SetLimit(int(b.cfg.General.DownloadConcurrency)) @@ -882,6 +886,24 @@ func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata. } } + // Try manifest-based download to avoid Walk (ListObjectsV2) + if manifest != nil { + manifestPrefix := path.Join("shadow", dbAndTableDir, capturedDisk, capturedPart.Name) + manifestFiles := manifest.FilesUnderPrefix(manifestPrefix) + if len(manifestFiles) > 0 { + pathSize, downloadErr := b.dst.DownloadPathWithManifest(dataCtx, partRemotePath, partLocalPath, manifestFiles, manifestPrefix, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond) + if downloadErr != nil { + return errors.WithMessage(downloadErr, "DownloadPathWithManifest") + } + atomic.AddUint64(&downloadedSize, uint64(pathSize)) + if b.resume { + b.resumableState.AppendToState(partRemotePath, pathSize) + } + log.Debug().Msgf("finish %s -> %s (manifest, %d files)", partRemotePath, partLocalPath, len(manifestFiles)) + return nil + } + } + // Fall back to Walk (ListObjectsV2) when no manifest is available pathSize, downloadErr := b.dst.DownloadPath(dataCtx, partRemotePath, partLocalPath, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond) if downloadErr != nil { return errors.WithMessage(downloadErr, "DownloadPath") diff --git a/pkg/backup/upload.go b/pkg/backup/upload.go index bf85f931..e0e48d51 100644 --- a/pkg/backup/upload.go +++ b/pkg/backup/upload.go @@ -148,6 +148,9 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr defer b.resumableState.Close() } + // Initialize file manifest to record all uploaded files for Walk-free restore + b.fileManifest = storage.NewBackupManifest(backupName) + compressedDataSize := int64(0) metadataSize := int64(0) @@ -272,6 +275,17 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr return errors.Wrapf(err, "can't upload %s", remoteBackupMetaFile) } } + // Record metadata.json in the manifest, then upload the manifest itself + b.recordUploadedFile(backupName, remoteBackupMetaFile, int64(len(newBackupMetadataBody))) + if b.fileManifest != nil { + if manifestErr := b.dst.UploadManifest(ctx, backupName, b.fileManifest); manifestErr != nil { + log.Warn().Err(manifestErr).Msg("failed to upload manifest.json, restore will fall back to Walk") + } else { + log.Info().Int("total_files", b.fileManifest.TotalFiles). + Int64("total_size", b.fileManifest.TotalSize). + Msg("uploaded backup manifest") + } + } log.Info().Fields(map[string]interface{}{ "backup": backupName, "operation": "upload", @@ -604,6 +618,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet } atomic.AddInt64(&uploadedBytes, uploadPathBytes) + b.recordUploadedLocalFiles(backupName, remotePath, backupPath, partFiles) if b.resume { b.resumableState.AppendToState(remotePathFull, uploadPathBytes) } @@ -649,6 +664,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet return errors.Wrapf(err, "can't check uploaded remoteDataFile: %s, error", remoteDataFile) } atomic.AddInt64(&uploadedBytes, remoteFile.Size()) + b.recordUploadedFile(backupName, remoteDataFile, remoteFile.Size()) if b.resume { b.resumableState.AppendToState(remoteDataFile, remoteFile.Size()) } @@ -707,6 +723,7 @@ func (b *Backuper) uploadTableMetadataRegular(ctx context.Context, backupName st if err != nil { return 0, errors.Wrap(err, "can't upload") } + b.recordUploadedFile(backupName, remoteTableMetaFile, int64(len(content))) if b.resume { b.resumableState.AppendToState(remoteTableMetaFile, int64(len(content))) } diff --git a/pkg/storage/manifest.go b/pkg/storage/manifest.go new file mode 100644 index 00000000..125c9915 --- /dev/null +++ b/pkg/storage/manifest.go @@ -0,0 +1,243 @@ +package storage + +import ( + "bytes" + "context" + "encoding/json" + "io" + "os" + "path" + "strings" + "time" + + "github.com/Altinity/clickhouse-backup/v2/pkg/common" + "github.com/eapache/go-resiliency/retrier" + "github.com/pkg/errors" + "github.com/rs/zerolog/log" +) + +const ( + // ManifestFileName is the name of the file manifest stored alongside metadata.json. + // It lists every file in the backup with its relative path, size, and last-modified time, + // allowing restore operations to skip the expensive S3 ListObjects Walk. + ManifestFileName = "manifest.json" + // ManifestVersion is the current manifest format version. + ManifestVersion = 1 +) + +// BackupManifest is a listing of all files within a single backup. +// It is written during upload and read during download to avoid +// the expensive recursive Walk (ListObjects) on object storage. +type BackupManifest struct { + Version int `json:"version"` + BackupName string `json:"backup_name"` + CreatedAt time.Time `json:"created_at"` + TotalSize int64 `json:"total_size"` + TotalFiles int `json:"total_files"` + Files []ManifestEntry `json:"files"` +} + +// ManifestEntry represents a single file in the backup manifest. +type ManifestEntry struct { + Path string `json:"path"` + Size int64 `json:"size"` + LastModified time.Time `json:"last_modified"` +} + +// manifestFile implements RemoteFile interface so manifest entries +// can be used directly in download paths that expect RemoteFile. +type manifestFile struct { + name string + size int64 + lastModified time.Time +} + +func (mf *manifestFile) Name() string { return mf.name } +func (mf *manifestFile) Size() int64 { return mf.size } +func (mf *manifestFile) LastModified() time.Time { return mf.lastModified } + +// ManifestEntryToRemoteFile converts a ManifestEntry to a RemoteFile, +// adjusting the path relative to the given prefix. +func ManifestEntryToRemoteFile(entry ManifestEntry, prefix string) RemoteFile { + name := entry.Path + if prefix != "" { + name = strings.TrimPrefix(name, prefix) + name = strings.TrimPrefix(name, "/") + } + return &manifestFile{ + name: name, + size: entry.Size, + lastModified: entry.LastModified, + } +} + +// NewBackupManifest creates a new empty manifest for the given backup. +func NewBackupManifest(backupName string) *BackupManifest { + return &BackupManifest{ + Version: ManifestVersion, + BackupName: backupName, + CreatedAt: time.Now().UTC(), + Files: make([]ManifestEntry, 0, 256), + } +} + +// NewBackupManifestWithCapacity creates a new empty manifest with a pre-allocated +// file slice capacity, reducing GC pressure for large backups by avoiding repeated +// slice growth. Use this when the expected file count is known or can be estimated. +func NewBackupManifestWithCapacity(backupName string, expectedFiles int) *BackupManifest { + if expectedFiles < 256 { + expectedFiles = 256 + } + return &BackupManifest{ + Version: ManifestVersion, + BackupName: backupName, + CreatedAt: time.Now().UTC(), + Files: make([]ManifestEntry, 0, expectedFiles), + } +} + +// AddFile adds a file entry to the manifest. +func (m *BackupManifest) AddFile(relativePath string, size int64, lastModified time.Time) { + m.Files = append(m.Files, ManifestEntry{ + Path: relativePath, + Size: size, + LastModified: lastModified, + }) + m.TotalFiles = len(m.Files) + m.TotalSize += size +} + +// HasFile returns true if the manifest contains an entry with the exact given path. +func (m *BackupManifest) HasFile(relativePath string) bool { + for _, f := range m.Files { + if f.Path == relativePath { + return true + } + } + return false +} + +// FilesUnderPrefix returns all manifest entries whose path starts with the given prefix. +// The prefix should NOT include the backup name (e.g., "shadow/default/my_table/default/part1"). +func (m *BackupManifest) FilesUnderPrefix(prefix string) []ManifestEntry { + prefix = strings.TrimSuffix(prefix, "/") + "/" + var result []ManifestEntry + for _, f := range m.Files { + if strings.HasPrefix(f.Path, prefix) { + result = append(result, f) + } + } + return result +} + +// Marshal serializes the manifest to JSON. +func (m *BackupManifest) Marshal() ([]byte, error) { + return json.MarshalIndent(m, "", "\t") +} + +// UnmarshalManifest deserializes a manifest from JSON. +func UnmarshalManifest(data []byte) (*BackupManifest, error) { + var m BackupManifest + if err := json.Unmarshal(data, &m); err != nil { + return nil, err + } + return &m, nil +} + +// UploadManifest uploads the manifest to remote storage alongside metadata.json. +func (bd *BackupDestination) UploadManifest(ctx context.Context, backupName string, manifest *BackupManifest) error { + data, err := manifest.Marshal() + if err != nil { + return err + } + remotePath := path.Join(backupName, ManifestFileName) + return bd.PutFile(ctx, remotePath, io.NopCloser(bytes.NewReader(data)), 0) +} + +// DownloadManifest attempts to download and parse the manifest for a given backup. +// Returns nil, nil if the manifest does not exist (graceful fallback to Walk). +func (bd *BackupDestination) DownloadManifest(ctx context.Context, backupName string) (*BackupManifest, error) { + remotePath := path.Join(backupName, ManifestFileName) + r, err := bd.GetFileReader(ctx, remotePath) + if err != nil { + // Manifest doesn't exist — this is expected for older backups + log.Debug().Str("backup", backupName).Msg("no manifest.json found, will fall back to Walk") + return nil, nil + } + defer r.Close() + data, err := io.ReadAll(r) + if err != nil { + log.Warn().Str("backup", backupName).Err(err).Msg("failed to read manifest.json, will fall back to Walk") + return nil, nil + } + manifest, err := UnmarshalManifest(data) + if err != nil { + log.Warn().Str("backup", backupName).Err(err).Msg("failed to parse manifest.json, will fall back to Walk") + return nil, nil + } + log.Info().Str("backup", backupName).Int("total_files", manifest.TotalFiles). + Int64("total_size", manifest.TotalSize).Msg("loaded backup manifest, skipping Walk") + return manifest, nil +} + +// DownloadPathWithManifest downloads files from remotePath using a pre-loaded manifest +// instead of calling Walk. Only files from manifestFiles are downloaded. +// prefixInManifest should be relative to the backup root (e.g., +// "shadow/default/my_table/default/part1"). +func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remotePath string, localPath string, manifestFiles []ManifestEntry, prefixInManifest string, RetriesOnFailure int, RetriesDuration time.Duration, RetriesJitter int8, RetrierClassifier retrier.Classifier, maxSpeed uint64) (int64, error) { + downloadedBytes := int64(0) + + for _, entry := range manifestFiles { + // Compute the relative name within remotePath + relativeName := strings.TrimPrefix(entry.Path, prefixInManifest) + relativeName = strings.TrimPrefix(relativeName, "/") + if relativeName == "" { + continue + } + f := ManifestEntryToRemoteFile(entry, prefixInManifest) + retry := retrier.New(retrier.ExponentialBackoff(RetriesOnFailure, common.AddRandomJitter(RetriesDuration, RetriesJitter)), RetrierClassifier) + err := retry.RunCtx(ctx, func(ctx context.Context) error { + startTime := time.Now() + r, err := bd.GetFileReader(ctx, path.Join(remotePath, f.Name())) + if err != nil { + log.Error().Err(err).Send() + return errors.WithMessage(err, "DownloadPathWithManifest GetFileReader") + } + dstFilePath := path.Join(localPath, f.Name()) + dstDirPath, _ := path.Split(dstFilePath) + if err := os.MkdirAll(dstDirPath, 0750); err != nil { + log.Error().Err(err).Send() + return errors.WithMessage(err, "DownloadPathWithManifest MkdirAll") + } + dst, err := os.Create(dstFilePath) + if err != nil { + log.Error().Err(err).Send() + return errors.WithMessage(err, "DownloadPathWithManifest Create") + } + if copyBytes, copyErr := bd.copyWithBuffer(dst, r); copyErr != nil { + log.Error().Err(copyErr).Send() + return errors.WithMessage(copyErr, "DownloadPathWithManifest io.Copy") + } else { + downloadedBytes += copyBytes + } + if dstCloseErr := dst.Close(); dstCloseErr != nil { + log.Error().Err(dstCloseErr).Send() + return errors.WithMessage(dstCloseErr, "DownloadPathWithManifest dst.Close") + } + if srcCloseErr := r.Close(); srcCloseErr != nil { + log.Error().Err(srcCloseErr).Send() + return errors.WithMessage(srcCloseErr, "DownloadPathWithManifest r.Close") + } + if dstFileInfo, statErr := os.Stat(dstFilePath); statErr == nil { + bd.throttleSpeed(startTime, dstFileInfo.Size(), maxSpeed) + } else { + return errors.WithMessage(statErr, "DownloadPathWithManifest Stat") + } + return nil + }) + if err != nil { + return downloadedBytes, errors.WithMessage(err, "DownloadPathWithManifest retry") + } + } + return downloadedBytes, nil +} diff --git a/pkg/storage/manifest_test.go b/pkg/storage/manifest_test.go new file mode 100644 index 00000000..c94b3369 --- /dev/null +++ b/pkg/storage/manifest_test.go @@ -0,0 +1,304 @@ +package storage + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewBackupManifest(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test_backup") + assert.Equal(t, ManifestVersion, m.Version) + assert.Equal(t, "test_backup", m.BackupName) + assert.NotZero(t, m.CreatedAt) + assert.Empty(t, m.Files) + assert.Equal(t, 0, m.TotalFiles) + assert.Equal(t, int64(0), m.TotalSize) +} + +func TestBackupManifest_AddFile(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test_backup") + now := time.Now().UTC() + + m.AddFile("shadow/default/table1/default/part1/data.bin", 12345, now) + m.AddFile("shadow/default/table1/default/part1/primary.idx", 456, now) + m.AddFile("metadata/default/table1.json", 789, now) + + assert.Equal(t, 3, m.TotalFiles) + assert.Equal(t, int64(12345+456+789), m.TotalSize) + assert.Len(t, m.Files, 3) + + assert.Equal(t, "shadow/default/table1/default/part1/data.bin", m.Files[0].Path) + assert.Equal(t, int64(12345), m.Files[0].Size) + assert.Equal(t, now, m.Files[0].LastModified) +} + +func TestBackupManifest_FilesUnderPrefix(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test_backup") + now := time.Now().UTC() + + // Add files from different tables and parts + m.AddFile("shadow/default/table1/default/part1/data.bin", 100, now) + m.AddFile("shadow/default/table1/default/part1/primary.idx", 50, now) + m.AddFile("shadow/default/table1/default/part2/data.bin", 200, now) + m.AddFile("shadow/default/table2/default/part1/data.bin", 300, now) + m.AddFile("metadata/default/table1.json", 10, now) + + // Files under part1 of table1 + files := m.FilesUnderPrefix("shadow/default/table1/default/part1") + assert.Len(t, files, 2) + for _, f := range files { + assert.Contains(t, f.Path, "shadow/default/table1/default/part1/") + } + + // Files under table1 (all parts) + files = m.FilesUnderPrefix("shadow/default/table1") + assert.Len(t, files, 3) // part1/data.bin, part1/primary.idx, part2/data.bin + + // Files under table2 + files = m.FilesUnderPrefix("shadow/default/table2") + assert.Len(t, files, 1) + + // Files under metadata + files = m.FilesUnderPrefix("metadata") + assert.Len(t, files, 1) + + // Non-existent prefix + files = m.FilesUnderPrefix("shadow/default/table3") + assert.Len(t, files, 0) +} + +func TestBackupManifest_FilesUnderPrefix_TrailingSlash(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test") + now := time.Now().UTC() + m.AddFile("shadow/db/tbl/disk/part1/file.bin", 100, now) + + // Should work with or without trailing slash + files1 := m.FilesUnderPrefix("shadow/db/tbl/disk/part1") + files2 := m.FilesUnderPrefix("shadow/db/tbl/disk/part1/") + assert.Equal(t, len(files1), len(files2)) +} + +func TestBackupManifest_MarshalUnmarshal(t *testing.T) { + t.Parallel() + m := NewBackupManifest("my_backup_2025") + now := time.Now().UTC().Truncate(time.Millisecond) // truncate for JSON roundtrip + + m.AddFile("shadow/default/orders/default/20250101_1_1_0/data.bin", 1048576, now) + m.AddFile("shadow/default/orders/default/20250101_1_1_0/primary.idx", 256, now) + m.AddFile("metadata/default/orders.json", 1024, now) + + data, err := m.Marshal() + require.NoError(t, err) + assert.Contains(t, string(data), `"version": 1`) + assert.Contains(t, string(data), `"backup_name": "my_backup_2025"`) + assert.Contains(t, string(data), `"total_files": 3`) + + // Unmarshal + m2, err := UnmarshalManifest(data) + require.NoError(t, err) + assert.Equal(t, m.Version, m2.Version) + assert.Equal(t, m.BackupName, m2.BackupName) + assert.Equal(t, m.TotalFiles, m2.TotalFiles) + assert.Equal(t, m.TotalSize, m2.TotalSize) + assert.Len(t, m2.Files, 3) + + // Verify file entries round-trip + assert.Equal(t, m.Files[0].Path, m2.Files[0].Path) + assert.Equal(t, m.Files[0].Size, m2.Files[0].Size) + assert.Equal(t, m.Files[1].Path, m2.Files[1].Path) + assert.Equal(t, m.Files[2].Path, m2.Files[2].Path) +} + +func TestUnmarshalManifest_InvalidJSON(t *testing.T) { + t.Parallel() + _, err := UnmarshalManifest([]byte(`not json`)) + assert.Error(t, err) +} + +func TestUnmarshalManifest_EmptyFiles(t *testing.T) { + t.Parallel() + data := []byte(`{"version":1,"backup_name":"empty","created_at":"2025-05-16T00:00:00Z","total_size":0,"total_files":0,"files":[]}`) + m, err := UnmarshalManifest(data) + require.NoError(t, err) + assert.Equal(t, "empty", m.BackupName) + assert.Empty(t, m.Files) +} + +func TestManifestEntryToRemoteFile(t *testing.T) { + t.Parallel() + now := time.Now().UTC() + entry := ManifestEntry{ + Path: "shadow/default/table1/default/part1/data.bin", + Size: 12345, + LastModified: now, + } + + // No prefix stripping + rf := ManifestEntryToRemoteFile(entry, "") + assert.Equal(t, "shadow/default/table1/default/part1/data.bin", rf.Name()) + assert.Equal(t, int64(12345), rf.Size()) + assert.Equal(t, now, rf.LastModified()) + + // With prefix stripping + rf2 := ManifestEntryToRemoteFile(entry, "shadow/default/table1/default/part1") + assert.Equal(t, "data.bin", rf2.Name()) + assert.Equal(t, int64(12345), rf2.Size()) +} + +func TestManifestEntryToRemoteFile_PrefixWithSlash(t *testing.T) { + t.Parallel() + entry := ManifestEntry{ + Path: "shadow/db/tbl/disk/part/subdir/file.bin", + Size: 100, + } + rf := ManifestEntryToRemoteFile(entry, "shadow/db/tbl/disk/part/") + assert.Equal(t, "subdir/file.bin", rf.Name()) +} + +func TestBackupManifest_LargeFileCount(t *testing.T) { + t.Parallel() + m := NewBackupManifest("large_backup") + now := time.Now().UTC() + + // Simulate a real backup with 10k files + for i := 0; i < 10000; i++ { + m.AddFile( + "shadow/default/trades/default/20250101_1_1_0/column"+string(rune('A'+i%26))+".bin", + int64(1024+i), + now, + ) + } + + assert.Equal(t, 10000, m.TotalFiles) + assert.Greater(t, m.TotalSize, int64(0)) + + // Marshal and unmarshal should work + data, err := m.Marshal() + require.NoError(t, err) + + m2, err := UnmarshalManifest(data) + require.NoError(t, err) + assert.Equal(t, 10000, m2.TotalFiles) + assert.Equal(t, m.TotalSize, m2.TotalSize) +} + +func TestManifestFileName_IsCorrect(t *testing.T) { + t.Parallel() + assert.Equal(t, "manifest.json", ManifestFileName) +} + +func TestManifestVersion_IsOne(t *testing.T) { + t.Parallel() + assert.Equal(t, 1, ManifestVersion) +} + +func TestBackupManifest_HasFile(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test") + now := time.Now().UTC() + + m.AddFile("metadata/default/orders.json", 100, now) + m.AddFile("shadow/default/orders/default/part1/data.bin", 200, now) + + assert.True(t, m.HasFile("metadata/default/orders.json")) + assert.True(t, m.HasFile("shadow/default/orders/default/part1/data.bin")) + assert.False(t, m.HasFile("metadata/default/nonexistent.json")) + assert.False(t, m.HasFile("metadata/default/orders.json/")) + assert.False(t, m.HasFile("")) +} + +func TestNewBackupManifestWithCapacity(t *testing.T) { + t.Parallel() + m := NewBackupManifestWithCapacity("test_backup", 5000) + assert.Equal(t, ManifestVersion, m.Version) + assert.Equal(t, "test_backup", m.BackupName) + assert.NotZero(t, m.CreatedAt) + assert.Empty(t, m.Files) + assert.Equal(t, 5000, cap(m.Files)) + assert.Equal(t, 0, m.TotalFiles) + assert.Equal(t, int64(0), m.TotalSize) +} + +func TestNewBackupManifestWithCapacity_MinimumFloor(t *testing.T) { + t.Parallel() + m := NewBackupManifestWithCapacity("test", 10) + assert.Equal(t, 256, cap(m.Files), "capacity should be at least 256") +} + +func TestNewBackupManifestWithCapacity_ZeroCapacity(t *testing.T) { + t.Parallel() + m := NewBackupManifestWithCapacity("test", 0) + assert.Equal(t, 256, cap(m.Files), "zero capacity should use floor of 256") +} + +func TestNewBackupManifestWithCapacity_NegativeCapacity(t *testing.T) { + t.Parallel() + m := NewBackupManifestWithCapacity("test", -100) + assert.Equal(t, 256, cap(m.Files), "negative capacity should use floor of 256") +} + +func TestNewBackupManifestWithCapacity_CorrectBehavior(t *testing.T) { + t.Parallel() + // Pre-sized manifest should produce identical results to default manifest + now := time.Now().UTC() + m1 := NewBackupManifest("test") + m2 := NewBackupManifestWithCapacity("test", 10000) + + for i := 0; i < 1000; i++ { + path := "shadow/default/table/default/part/col" + string(rune('A'+i%26)) + ".bin" + m1.AddFile(path, int64(i*100), now) + m2.AddFile(path, int64(i*100), now) + } + + assert.Equal(t, m1.TotalFiles, m2.TotalFiles) + assert.Equal(t, m1.TotalSize, m2.TotalSize) + assert.Equal(t, len(m1.Files), len(m2.Files)) + + // Verify marshal/unmarshal produces identical output + data1, err1 := m1.Marshal() + data2, err2 := m2.Marshal() + require.NoError(t, err1) + require.NoError(t, err2) + + // Unmarshal and compare (can't compare JSON directly due to timestamp) + um1, _ := UnmarshalManifest(data1) + um2, _ := UnmarshalManifest(data2) + assert.Equal(t, um1.TotalFiles, um2.TotalFiles) + assert.Equal(t, um1.TotalSize, um2.TotalSize) +} + +func TestNewBackupManifestWithCapacity_LargeCapacityNoAlloc(t *testing.T) { + t.Parallel() + // Verify that pre-sizing to exact capacity produces correct results + m := NewBackupManifestWithCapacity("test", 50000) + now := time.Now().UTC() + + for i := 0; i < 50000; i++ { + m.AddFile("shadow/default/t/d/p/f.bin", int64(i), now) + } + assert.Equal(t, 50000, m.TotalFiles) + // Pre-sized capacity should prevent any reallocation + assert.GreaterOrEqual(t, cap(m.Files), 50000) +} + +func TestBackupManifest_FilesUnderPrefix_NoPartialMatch(t *testing.T) { + t.Parallel() + m := NewBackupManifest("test") + now := time.Now().UTC() + + // "part1" should not match "part10" or "part1_suffix" + m.AddFile("shadow/db/tbl/disk/part1/file.bin", 100, now) + m.AddFile("shadow/db/tbl/disk/part10/file.bin", 200, now) + m.AddFile("shadow/db/tbl/disk/part1_suffix/file.bin", 300, now) + + files := m.FilesUnderPrefix("shadow/db/tbl/disk/part1") + assert.Len(t, files, 1, "should only match exact prefix with trailing slash") + assert.Equal(t, "shadow/db/tbl/disk/part1/file.bin", files[0].Path) +} From 977113d2562141f9a768e8de82b35b67b1a99fac Mon Sep 17 00:00:00 2001 From: slach Date: Tue, 7 Jul 2026 09:47:30 +0400 Subject: [PATCH 2/5] fixes after merge master --- pkg/storage/manifest.go | 32 +++++++++++++++++++++++--------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/pkg/storage/manifest.go b/pkg/storage/manifest.go index 125c9915..798d9f68 100644 --- a/pkg/storage/manifest.go +++ b/pkg/storage/manifest.go @@ -8,9 +8,11 @@ import ( "os" "path" "strings" + "sync" "time" "github.com/Altinity/clickhouse-backup/v2/pkg/common" + "github.com/Altinity/clickhouse-backup/v2/pkg/storage/bwlimit" "github.com/eapache/go-resiliency/retrier" "github.com/pkg/errors" "github.com/rs/zerolog/log" @@ -52,8 +54,8 @@ type manifestFile struct { lastModified time.Time } -func (mf *manifestFile) Name() string { return mf.name } -func (mf *manifestFile) Size() int64 { return mf.size } +func (mf *manifestFile) Name() string { return mf.name } +func (mf *manifestFile) Size() int64 { return mf.size } func (mf *manifestFile) LastModified() time.Time { return mf.lastModified } // ManifestEntryToRemoteFile converts a ManifestEntry to a RemoteFile, @@ -186,6 +188,7 @@ func (bd *BackupDestination) DownloadManifest(ctx context.Context, backupName st // "shadow/default/my_table/default/part1"). func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remotePath string, localPath string, manifestFiles []ManifestEntry, prefixInManifest string, RetriesOnFailure int, RetriesDuration time.Duration, RetriesJitter int8, RetrierClassifier retrier.Classifier, maxSpeed uint64) (int64, error) { downloadedBytes := int64(0) + limiter := bd.DownloadLimiter(maxSpeed) for _, entry := range manifestFiles { // Compute the relative name within remotePath @@ -197,12 +200,28 @@ func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remot f := ManifestEntryToRemoteFile(entry, prefixInManifest) retry := retrier.New(retrier.ExponentialBackoff(RetriesOnFailure, common.AddRandomJitter(RetriesDuration, RetriesJitter)), RetrierClassifier) err := retry.RunCtx(ctx, func(ctx context.Context) error { - startTime := time.Now() r, err := bd.GetFileReader(ctx, path.Join(remotePath, f.Name())) if err != nil { log.Error().Err(err).Send() return errors.WithMessage(err, "DownloadPathWithManifest GetFileReader") } + r = bwlimit.ReadCloser(ctx, r, limiter) + var closeSrcOnce sync.Once + var srcCloseErr error + closeSrc := func() { closeSrcOnce.Do(func() { srcCloseErr = r.Close() }) } + // A stalled read is not interruptible by context alone: copyWithBuffer + // blocks in Read and never re-checks ctx, so /backup/kill cancels the + // context but the copy keeps running. Force-close the source reader on + // cancellation so the blocked read returns and the copy unwinds. + watchDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + closeSrc() + case <-watchDone: + } + }() + defer close(watchDone) dstFilePath := path.Join(localPath, f.Name()) dstDirPath, _ := path.Split(dstFilePath) if err := os.MkdirAll(dstDirPath, 0750); err != nil { @@ -224,15 +243,10 @@ func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remot log.Error().Err(dstCloseErr).Send() return errors.WithMessage(dstCloseErr, "DownloadPathWithManifest dst.Close") } - if srcCloseErr := r.Close(); srcCloseErr != nil { + if closeSrc(); srcCloseErr != nil { log.Error().Err(srcCloseErr).Send() return errors.WithMessage(srcCloseErr, "DownloadPathWithManifest r.Close") } - if dstFileInfo, statErr := os.Stat(dstFilePath); statErr == nil { - bd.throttleSpeed(startTime, dstFileInfo.Size(), maxSpeed) - } else { - return errors.WithMessage(statErr, "DownloadPathWithManifest Stat") - } return nil }) if err != nil { From f3df4c24a5a9ac6404c2930600b6d6cfc336c4d5 Mon Sep 17 00:00:00 2001 From: slach Date: Tue, 7 Jul 2026 11:39:12 +0400 Subject: [PATCH 3/5] replace manifest.json with gzip-compressed bbolt manifest.bolt.gz The JSON manifest didn't scale to backups with millions of files: indented uncompressed JSON grew to hundreds of MB in the bucket, was fully loaded into memory on both upload and restore, and FilesUnderPrefix was a linear scan per part, O(parts x files). Rework the manifest as a bbolt database (same library as resumable .state2), gzip-compressed for transfer: - upload: ManifestWriter batches file paths into a local temp bolt file, 10k keys per transaction, so memory is bounded by batch size - restore: ManifestReader opens the decompressed bolt file read-only (mmap), per-part lookup is a sorted-key prefix seek instead of a full scan - store only file paths as keys: size and mtime were never read on the restore path, dropping them also removes the per-file os.Stat during upload - keep graceful fallback to Walk when the manifest is missing or unreadable, and keep manifest failures non-fatal for the backup --- pkg/backup/backuper.go | 30 +-- pkg/backup/download.go | 13 +- pkg/backup/upload.go | 21 +- pkg/storage/manifest.go | 444 ++++++++++++++++++++++++----------- pkg/storage/manifest_test.go | 384 ++++++++++++------------------ 5 files changed, 487 insertions(+), 405 deletions(-) diff --git a/pkg/backup/backuper.go b/pkg/backup/backuper.go index ec2a263a..044c3a16 100644 --- a/pkg/backup/backuper.go +++ b/pkg/backup/backuper.go @@ -11,7 +11,6 @@ import ( "regexp" "strings" "sync" - "time" "github.com/Altinity/clickhouse-backup/v2/pkg/common" "github.com/Altinity/clickhouse-backup/v2/pkg/metadata" @@ -52,8 +51,7 @@ type Backuper struct { resumableState *resumable.State shadowBackupUUIDs []string shadowBackupUUIDsMutex sync.Mutex - fileManifest *storage.BackupManifest - fileManifestMu sync.Mutex + fileManifest *storage.ManifestWriter } func NewBackuper(cfg *config.Config, opts ...BackuperOpt) *Backuper { @@ -592,34 +590,22 @@ func (b *Backuper) addShadowBackupUUID(uuid string) { // recordUploadedFile records a single file in the backup manifest (thread-safe). // remotePath should be the full remote path including the backupName prefix. -func (b *Backuper) recordUploadedFile(backupName, remotePath string, size int64) { +func (b *Backuper) recordUploadedFile(backupName, remotePath string) { if b.fileManifest == nil { return } - relPath := strings.TrimPrefix(remotePath, backupName+"/") - b.fileManifestMu.Lock() - b.fileManifest.AddFile(relPath, size, time.Now().UTC()) - b.fileManifestMu.Unlock() + b.fileManifest.AddFile(strings.TrimPrefix(remotePath, backupName+"/")) } -// recordUploadedLocalFiles records multiple files in the backup manifest by -// stating the local source files to obtain their sizes (thread-safe). -// remotePath is the remote directory (including backupName prefix), localBasePath -// is the local directory the files are relative to. -func (b *Backuper) recordUploadedLocalFiles(backupName, remotePath, localBasePath string, files []string) { +// recordUploadedFiles records multiple files in the backup manifest (thread-safe). +// remotePath is the remote directory (including backupName prefix), files are +// relative to it. +func (b *Backuper) recordUploadedFiles(backupName, remotePath string, files []string) { if b.fileManifest == nil { return } - b.fileManifestMu.Lock() - defer b.fileManifestMu.Unlock() for _, f := range files { - fullLocal := path.Join(localBasePath, f) - info, err := os.Stat(fullLocal) - if err != nil { - continue - } - relPath := strings.TrimPrefix(path.Join(remotePath, f), backupName+"/") - b.fileManifest.AddFile(relPath, info.Size(), info.ModTime()) + b.fileManifest.AddFile(strings.TrimPrefix(path.Join(remotePath, f), backupName+"/")) } } diff --git a/pkg/backup/download.go b/pkg/backup/download.go index 28944145..7184fff9 100644 --- a/pkg/backup/download.go +++ b/pkg/backup/download.go @@ -166,8 +166,8 @@ func (b *Backuper) Download(backupName string, tablePattern string, partitions [ return errors.Errorf("'%s' is not found on remote storage", backupName) } // Download file manifest for Walk-free restore (falls back gracefully if not present) - var backupManifest *storage.BackupManifest - backupManifest, _ = b.dst.DownloadManifest(ctx, backupName) + backupManifest := b.dst.DownloadManifest(ctx, backupName) + defer backupManifest.Close() if len(remoteBackup.Tables) == 0 && remoteBackup.RBACSize == 0 && remoteBackup.ConfigSize == 0 && remoteBackup.NamedCollectionsSize == 0 && !b.cfg.General.AllowEmptyBackups { return errors.Errorf("'%s' is empty backup", backupName) @@ -744,7 +744,7 @@ func (b *Backuper) downloadBackupRelatedDir(ctx context.Context, remoteBackup st return uint64(remoteFileInfo.Size()), nil } -func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.BackupMetadata, table metadata.TableMetadata, disks []clickhouse.Disk, hardlinkExistsFiles bool, manifest *storage.BackupManifest) (uint64, error) { +func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata.BackupMetadata, table metadata.TableMetadata, disks []clickhouse.Disk, hardlinkExistsFiles bool, manifest *storage.ManifestReader) (uint64, error) { dbAndTableDir := path.Join(common.TablePathEncode(table.Database), common.TablePathEncode(table.Table)) dataGroup, dataCtx := errgroup.WithContext(ctx) dataGroup.SetLimit(int(b.cfg.General.DownloadConcurrency)) @@ -922,9 +922,12 @@ func (b *Backuper) downloadTableData(ctx context.Context, remoteBackup metadata. // Try manifest-based download to avoid Walk (ListObjectsV2) if manifest != nil { manifestPrefix := path.Join("shadow", dbAndTableDir, capturedDisk, capturedPart.Name) - manifestFiles := manifest.FilesUnderPrefix(manifestPrefix) + manifestFiles, manifestErr := manifest.FilesUnderPrefix(manifestPrefix) + if manifestErr != nil { + log.Warn().Err(manifestErr).Msgf("manifest lookup failed for %s, will fall back to Walk", manifestPrefix) + } if len(manifestFiles) > 0 { - pathSize, downloadErr := b.dst.DownloadPathWithManifest(dataCtx, partRemotePath, partLocalPath, manifestFiles, manifestPrefix, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond) + pathSize, downloadErr := b.dst.DownloadPathWithManifest(dataCtx, partRemotePath, partLocalPath, manifestFiles, b.cfg.General.RetriesOnFailure, b.cfg.General.RetriesDuration, b.cfg.General.RetriesJitter, b, b.cfg.General.DownloadMaxBytesPerSecond) if downloadErr != nil { return errors.WithMessage(downloadErr, "DownloadPathWithManifest") } diff --git a/pkg/backup/upload.go b/pkg/backup/upload.go index 78cd0419..cabd849c 100644 --- a/pkg/backup/upload.go +++ b/pkg/backup/upload.go @@ -149,7 +149,15 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr } // Initialize file manifest to record all uploaded files for Walk-free restore - b.fileManifest = storage.NewBackupManifest(backupName) + if manifestWriter, manifestErr := storage.NewManifestWriter(backupName); manifestErr != nil { + log.Warn().Err(manifestErr).Msgf("can't create %s writer, restore will fall back to Walk", storage.ManifestFileName) + } else { + b.fileManifest = manifestWriter + } + defer func() { + b.fileManifest.Close() + b.fileManifest = nil + }() compressedDataSize := int64(0) metadataSize := int64(0) @@ -282,13 +290,12 @@ func (b *Backuper) Upload(backupName string, deleteSource bool, diffFrom, diffFr } } // Record metadata.json in the manifest, then upload the manifest itself - b.recordUploadedFile(backupName, remoteBackupMetaFile, int64(len(newBackupMetadataBody))) + b.recordUploadedFile(backupName, remoteBackupMetaFile) if b.fileManifest != nil { if manifestErr := b.dst.UploadManifest(ctx, backupName, b.fileManifest); manifestErr != nil { - log.Warn().Err(manifestErr).Msg("failed to upload manifest.json, restore will fall back to Walk") + log.Warn().Err(manifestErr).Msgf("failed to upload %s, restore will fall back to Walk", storage.ManifestFileName) } else { log.Info().Int("total_files", b.fileManifest.TotalFiles). - Int64("total_size", b.fileManifest.TotalSize). Msg("uploaded backup manifest") } } @@ -642,7 +649,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet } atomic.AddInt64(&uploadedBytes, uploadPathBytes) - b.recordUploadedLocalFiles(backupName, remotePath, backupPath, partFiles) + b.recordUploadedFiles(backupName, remotePath, partFiles) if b.resume { if err = b.resumableState.AppendToState(remotePathFull, uploadPathBytes); err != nil { return errors.Wrap(err, "resumableState.AppendToState") @@ -694,7 +701,7 @@ func (b *Backuper) uploadTableData(ctx context.Context, backupName string, delet return errors.Wrapf(err, "can't check uploaded remoteDataFile: %s, error", remoteDataFile) } atomic.AddInt64(&uploadedBytes, remoteFile.Size()) - b.recordUploadedFile(backupName, remoteDataFile, remoteFile.Size()) + b.recordUploadedFile(backupName, remoteDataFile) if b.resume { if err = b.resumableState.AppendToState(remoteDataFile, remoteFile.Size()); err != nil { return errors.Wrap(err, "resumableState.AppendToState") @@ -759,7 +766,7 @@ func (b *Backuper) uploadTableMetadataRegular(ctx context.Context, backupName st if err != nil { return 0, errors.Wrap(err, "can't upload") } - b.recordUploadedFile(backupName, remoteTableMetaFile, int64(len(content))) + b.recordUploadedFile(backupName, remoteTableMetaFile) if b.resume { if err = b.resumableState.AppendToState(remoteTableMetaFile, int64(len(content))); err != nil { return 0, errors.Wrap(err, "resumableState.AppendToState") diff --git a/pkg/storage/manifest.go b/pkg/storage/manifest.go index 798d9f68..efc2da40 100644 --- a/pkg/storage/manifest.go +++ b/pkg/storage/manifest.go @@ -2,8 +2,9 @@ package storage import ( "bytes" + "compress/gzip" "context" - "encoding/json" + "encoding/binary" "io" "os" "path" @@ -16,191 +17,368 @@ import ( "github.com/eapache/go-resiliency/retrier" "github.com/pkg/errors" "github.com/rs/zerolog/log" + bolt "go.etcd.io/bbolt" ) const ( // ManifestFileName is the name of the file manifest stored alongside metadata.json. - // It lists every file in the backup with its relative path, size, and last-modified time, - // allowing restore operations to skip the expensive S3 ListObjects Walk. - ManifestFileName = "manifest.json" + // It is a gzip-compressed bbolt database whose keys are the relative paths of every + // file in the backup, allowing restore operations to skip the expensive S3 + // ListObjects Walk. bbolt keeps keys sorted, so per-part lookups are prefix seeks + // instead of full scans, and the read side is mmap-backed instead of loading the + // whole listing into the heap. + ManifestFileName = "manifest.bolt.gz" // ManifestVersion is the current manifest format version. ManifestVersion = 1 + // manifestBatchSize is how many recorded files are buffered in memory + // before being flushed to the local bolt database in one transaction. + manifestBatchSize = 10000 ) -// BackupManifest is a listing of all files within a single backup. -// It is written during upload and read during download to avoid -// the expensive recursive Walk (ListObjects) on object storage. -type BackupManifest struct { - Version int `json:"version"` - BackupName string `json:"backup_name"` - CreatedAt time.Time `json:"created_at"` - TotalSize int64 `json:"total_size"` - TotalFiles int `json:"total_files"` - Files []ManifestEntry `json:"files"` -} - -// ManifestEntry represents a single file in the backup manifest. -type ManifestEntry struct { - Path string `json:"path"` - Size int64 `json:"size"` - LastModified time.Time `json:"last_modified"` -} +var ( + manifestFilesBucket = []byte("files") + manifestMetaBucket = []byte("meta") +) -// manifestFile implements RemoteFile interface so manifest entries -// can be used directly in download paths that expect RemoteFile. -type manifestFile struct { - name string - size int64 - lastModified time.Time +// ManifestWriter incrementally records uploaded file paths into a local bbolt +// database, batching writes to bound memory usage. It is safe for concurrent +// use. After a write error it becomes a no-op and the error is returned by +// Finalize, so a broken manifest never fails the backup itself. +type ManifestWriter struct { + mu sync.Mutex + db *bolt.DB + localPath string + gzPath string + backupName string + createdAt time.Time + batch []string + err error + TotalFiles int } -func (mf *manifestFile) Name() string { return mf.name } -func (mf *manifestFile) Size() int64 { return mf.size } -func (mf *manifestFile) LastModified() time.Time { return mf.lastModified } - -// ManifestEntryToRemoteFile converts a ManifestEntry to a RemoteFile, -// adjusting the path relative to the given prefix. -func ManifestEntryToRemoteFile(entry ManifestEntry, prefix string) RemoteFile { - name := entry.Path - if prefix != "" { - name = strings.TrimPrefix(name, prefix) - name = strings.TrimPrefix(name, "/") - } - return &manifestFile{ - name: name, - size: entry.Size, - lastModified: entry.LastModified, +// NewManifestWriter creates a manifest writer backed by a temporary local bolt file. +func NewManifestWriter(backupName string) (*ManifestWriter, error) { + tmpFile, err := os.CreateTemp("", "clickhouse-backup-manifest-*.bolt") + if err != nil { + return nil, errors.WithMessage(err, "manifest: can't create temporary file") } -} - -// NewBackupManifest creates a new empty manifest for the given backup. -func NewBackupManifest(backupName string) *BackupManifest { - return &BackupManifest{ - Version: ManifestVersion, - BackupName: backupName, - CreatedAt: time.Now().UTC(), - Files: make([]ManifestEntry, 0, 256), + localPath := tmpFile.Name() + if err = tmpFile.Close(); err != nil { + _ = os.Remove(localPath) + return nil, errors.WithMessage(err, "manifest: can't close temporary file") } -} - -// NewBackupManifestWithCapacity creates a new empty manifest with a pre-allocated -// file slice capacity, reducing GC pressure for large backups by avoiding repeated -// slice growth. Use this when the expected file count is known or can be estimated. -func NewBackupManifestWithCapacity(backupName string, expectedFiles int) *BackupManifest { - if expectedFiles < 256 { - expectedFiles = 256 - } - return &BackupManifest{ - Version: ManifestVersion, - BackupName: backupName, - CreatedAt: time.Now().UTC(), - Files: make([]ManifestEntry, 0, expectedFiles), + db, err := bolt.Open(localPath, 0600, nil) + if err != nil { + _ = os.Remove(localPath) + return nil, errors.WithMessage(err, "manifest: can't open bolt database") + } + err = db.Update(func(tx *bolt.Tx) error { + if _, createErr := tx.CreateBucketIfNotExists(manifestFilesBucket); createErr != nil { + return createErr + } + _, createErr := tx.CreateBucketIfNotExists(manifestMetaBucket) + return createErr + }) + if err != nil { + _ = db.Close() + _ = os.Remove(localPath) + return nil, errors.WithMessage(err, "manifest: can't create buckets") } + return &ManifestWriter{ + db: db, + localPath: localPath, + backupName: backupName, + createdAt: time.Now().UTC(), + }, nil } -// AddFile adds a file entry to the manifest. -func (m *BackupManifest) AddFile(relativePath string, size int64, lastModified time.Time) { - m.Files = append(m.Files, ManifestEntry{ - Path: relativePath, - Size: size, - LastModified: lastModified, - }) - m.TotalFiles = len(m.Files) - m.TotalSize += size +// AddFile records a file path in the manifest. On flush failure the writer is +// marked broken and subsequent calls become no-ops; the error surfaces in Finalize. +func (w *ManifestWriter) AddFile(relativePath string) { + w.mu.Lock() + defer w.mu.Unlock() + if w.err != nil || w.db == nil { + return + } + w.batch = append(w.batch, relativePath) + w.TotalFiles++ + if len(w.batch) >= manifestBatchSize { + if flushErr := w.flushLocked(); flushErr != nil { + w.err = flushErr + log.Warn().Err(flushErr).Msgf("manifest: flush failed, %s will not be uploaded", ManifestFileName) + } + } } -// HasFile returns true if the manifest contains an entry with the exact given path. -func (m *BackupManifest) HasFile(relativePath string) bool { - for _, f := range m.Files { - if f.Path == relativePath { - return true +func (w *ManifestWriter) flushLocked() error { + if len(w.batch) == 0 { + return nil + } + err := w.db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket(manifestFilesBucket) + for _, p := range w.batch { + if putErr := b.Put([]byte(p), []byte{}); putErr != nil { + return putErr + } } + return nil + }) + if err == nil { + w.batch = w.batch[:0] } - return false + return err } -// FilesUnderPrefix returns all manifest entries whose path starts with the given prefix. -// The prefix should NOT include the backup name (e.g., "shadow/default/my_table/default/part1"). -func (m *BackupManifest) FilesUnderPrefix(prefix string) []ManifestEntry { - prefix = strings.TrimSuffix(prefix, "/") + "/" - var result []ManifestEntry - for _, f := range m.Files { - if strings.HasPrefix(f.Path, prefix) { - result = append(result, f) +// Finalize flushes pending entries, writes the manifest metadata, closes the +// bolt database and gzip-compresses it. It returns the path of the compressed file. +func (w *ManifestWriter) Finalize() (string, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.err != nil { + return "", w.err + } + if w.db == nil { + return "", errors.New("manifest: writer already finalized") + } + if flushErr := w.flushLocked(); flushErr != nil { + w.err = flushErr + return "", flushErr + } + metaErr := w.db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket(manifestMetaBucket) + buf := make([]byte, binary.MaxVarintLen64) + n := binary.PutVarint(buf, int64(ManifestVersion)) + if err := b.Put([]byte("version"), append([]byte(nil), buf[:n]...)); err != nil { + return err } + if err := b.Put([]byte("backup_name"), []byte(w.backupName)); err != nil { + return err + } + if err := b.Put([]byte("created_at"), []byte(w.createdAt.Format(time.RFC3339))); err != nil { + return err + } + n = binary.PutVarint(buf, int64(w.TotalFiles)) + return b.Put([]byte("total_files"), append([]byte(nil), buf[:n]...)) + }) + if metaErr != nil { + w.err = metaErr + return "", metaErr + } + if closeErr := w.db.Close(); closeErr != nil { + w.err = closeErr + return "", closeErr + } + w.db = nil + gzPath, gzErr := gzipFile(w.localPath) + if gzErr != nil { + w.err = gzErr + return "", gzErr } - return result + w.gzPath = gzPath + return gzPath, nil } -// Marshal serializes the manifest to JSON. -func (m *BackupManifest) Marshal() ([]byte, error) { - return json.MarshalIndent(m, "", "\t") +func gzipFile(srcPath string) (string, error) { + src, err := os.Open(srcPath) + if err != nil { + return "", errors.WithMessage(err, "manifest: can't open bolt file for compression") + } + defer src.Close() + gzPath := srcPath + ".gz" + dst, err := os.Create(gzPath) + if err != nil { + return "", errors.WithMessage(err, "manifest: can't create gzip file") + } + gz := gzip.NewWriter(dst) + if _, err = io.Copy(gz, src); err != nil { + _ = dst.Close() + _ = os.Remove(gzPath) + return "", errors.WithMessage(err, "manifest: can't compress bolt file") + } + if err = gz.Close(); err != nil { + _ = dst.Close() + _ = os.Remove(gzPath) + return "", errors.WithMessage(err, "manifest: can't finish compression") + } + if err = dst.Close(); err != nil { + _ = os.Remove(gzPath) + return "", errors.WithMessage(err, "manifest: can't close gzip file") + } + return gzPath, nil } -// UnmarshalManifest deserializes a manifest from JSON. -func UnmarshalManifest(data []byte) (*BackupManifest, error) { - var m BackupManifest - if err := json.Unmarshal(data, &m); err != nil { - return nil, err +// Close releases the writer's resources and removes its temporary files. +// Safe to call multiple times, after Finalize, and on a nil writer. +func (w *ManifestWriter) Close() { + if w == nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if w.db != nil { + if err := w.db.Close(); err != nil { + log.Warn().Err(err).Msgf("manifest: can't close %s", w.localPath) + } + w.db = nil + } + if w.localPath != "" { + _ = os.Remove(w.localPath) + w.localPath = "" + } + if w.gzPath != "" { + _ = os.Remove(w.gzPath) + w.gzPath = "" } - return &m, nil } -// UploadManifest uploads the manifest to remote storage alongside metadata.json. -func (bd *BackupDestination) UploadManifest(ctx context.Context, backupName string, manifest *BackupManifest) error { - data, err := manifest.Marshal() +// UploadManifest finalizes the manifest and uploads it to remote storage +// alongside metadata.json. +func (bd *BackupDestination) UploadManifest(ctx context.Context, backupName string, w *ManifestWriter) error { + gzPath, err := w.Finalize() if err != nil { return err } + f, err := os.Open(gzPath) + if err != nil { + return errors.WithMessage(err, "manifest: can't open compressed manifest") + } + fInfo, err := f.Stat() + if err != nil { + _ = f.Close() + return errors.WithMessage(err, "manifest: can't stat compressed manifest") + } remotePath := path.Join(backupName, ManifestFileName) - return bd.PutFile(ctx, remotePath, io.NopCloser(bytes.NewReader(data)), 0) + return bd.PutFile(ctx, remotePath, f, fInfo.Size()) } -// DownloadManifest attempts to download and parse the manifest for a given backup. -// Returns nil, nil if the manifest does not exist (graceful fallback to Walk). -func (bd *BackupDestination) DownloadManifest(ctx context.Context, backupName string) (*BackupManifest, error) { +// ManifestReader provides prefix lookups over a downloaded manifest. It is +// backed by a read-only (mmap) bolt database in a local temporary file, so +// lookups don't require loading the whole listing into memory. +type ManifestReader struct { + db *bolt.DB + localPath string +} + +// DownloadManifest attempts to download, decompress and open the manifest for +// a given backup. Returns nil if the manifest does not exist or is unreadable +// (graceful fallback to Walk). +func (bd *BackupDestination) DownloadManifest(ctx context.Context, backupName string) *ManifestReader { remotePath := path.Join(backupName, ManifestFileName) r, err := bd.GetFileReader(ctx, remotePath) if err != nil { // Manifest doesn't exist — this is expected for older backups - log.Debug().Str("backup", backupName).Msg("no manifest.json found, will fall back to Walk") - return nil, nil + log.Debug().Str("backup", backupName).Msgf("no %s found, will fall back to Walk", ManifestFileName) + return nil } defer r.Close() - data, err := io.ReadAll(r) + gz, err := gzip.NewReader(r) + if err != nil { + log.Warn().Str("backup", backupName).Err(err).Msgf("can't decompress %s, will fall back to Walk", ManifestFileName) + return nil + } + tmpFile, err := os.CreateTemp("", "clickhouse-backup-manifest-*.bolt") if err != nil { - log.Warn().Str("backup", backupName).Err(err).Msg("failed to read manifest.json, will fall back to Walk") - return nil, nil + log.Warn().Str("backup", backupName).Err(err).Msgf("can't create temporary file for %s, will fall back to Walk", ManifestFileName) + return nil + } + tmpPath := tmpFile.Name() + if _, err = io.Copy(tmpFile, gz); err != nil { + _ = tmpFile.Close() + _ = os.Remove(tmpPath) + log.Warn().Str("backup", backupName).Err(err).Msgf("can't read %s, will fall back to Walk", ManifestFileName) + return nil + } + if err = tmpFile.Close(); err != nil { + _ = os.Remove(tmpPath) + log.Warn().Str("backup", backupName).Err(err).Msgf("can't write %s, will fall back to Walk", ManifestFileName) + return nil } - manifest, err := UnmarshalManifest(data) + reader, err := openManifestReader(tmpPath) if err != nil { - log.Warn().Str("backup", backupName).Err(err).Msg("failed to parse manifest.json, will fall back to Walk") - return nil, nil + _ = os.Remove(tmpPath) + log.Warn().Str("backup", backupName).Err(err).Msgf("can't open %s, will fall back to Walk", ManifestFileName) + return nil } - log.Info().Str("backup", backupName).Int("total_files", manifest.TotalFiles). - Int64("total_size", manifest.TotalSize).Msg("loaded backup manifest, skipping Walk") - return manifest, nil + return reader } -// DownloadPathWithManifest downloads files from remotePath using a pre-loaded manifest -// instead of calling Walk. Only files from manifestFiles are downloaded. -// prefixInManifest should be relative to the backup root (e.g., -// "shadow/default/my_table/default/part1"). -func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remotePath string, localPath string, manifestFiles []ManifestEntry, prefixInManifest string, RetriesOnFailure int, RetriesDuration time.Duration, RetriesJitter int8, RetrierClassifier retrier.Classifier, maxSpeed uint64) (int64, error) { +func openManifestReader(localPath string) (*ManifestReader, error) { + db, err := bolt.Open(localPath, 0600, &bolt.Options{ReadOnly: true}) + if err != nil { + return nil, errors.WithMessage(err, "manifest: can't open bolt database") + } + err = db.View(func(tx *bolt.Tx) error { + if tx.Bucket(manifestFilesBucket) == nil || tx.Bucket(manifestMetaBucket) == nil { + return errors.New("manifest: missing buckets") + } + versionBytes := tx.Bucket(manifestMetaBucket).Get([]byte("version")) + if versionBytes == nil { + return errors.New("manifest: missing version") + } + version, n := binary.Varint(versionBytes) + if n <= 0 || version != ManifestVersion { + return errors.Errorf("manifest: unsupported version %d", version) + } + return nil + }) + if err != nil { + _ = db.Close() + return nil, err + } + return &ManifestReader{db: db, localPath: localPath}, nil +} + +// FilesUnderPrefix returns the names (relative to the prefix) of all manifest +// entries whose path starts with the given prefix, using a sorted-key seek +// instead of a full scan. The prefix should NOT include the backup name +// (e.g., "shadow/default/my_table/default/part1"). +func (r *ManifestReader) FilesUnderPrefix(prefix string) ([]string, error) { + prefixBytes := []byte(strings.TrimSuffix(prefix, "/") + "/") + var result []string + err := r.db.View(func(tx *bolt.Tx) error { + c := tx.Bucket(manifestFilesBucket).Cursor() + for k, _ := c.Seek(prefixBytes); k != nil && bytes.HasPrefix(k, prefixBytes); k, _ = c.Next() { + if name := string(k[len(prefixBytes):]); name != "" { + result = append(result, name) + } + } + return nil + }) + if err != nil { + return nil, err + } + return result, nil +} + +// Close closes the underlying bolt database and removes the temporary file. +// Safe to call on a nil reader. +func (r *ManifestReader) Close() { + if r == nil { + return + } + if r.db != nil { + if err := r.db.Close(); err != nil { + log.Warn().Err(err).Msgf("manifest: can't close %s", r.localPath) + } + r.db = nil + } + if r.localPath != "" { + _ = os.Remove(r.localPath) + r.localPath = "" + } +} + +// DownloadPathWithManifest downloads the given files from remotePath using a +// pre-loaded manifest listing instead of calling Walk. fileNames are relative +// to remotePath, as returned by ManifestReader.FilesUnderPrefix. +func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remotePath string, localPath string, fileNames []string, RetriesOnFailure int, RetriesDuration time.Duration, RetriesJitter int8, RetrierClassifier retrier.Classifier, maxSpeed uint64) (int64, error) { downloadedBytes := int64(0) limiter := bd.DownloadLimiter(maxSpeed) - for _, entry := range manifestFiles { - // Compute the relative name within remotePath - relativeName := strings.TrimPrefix(entry.Path, prefixInManifest) - relativeName = strings.TrimPrefix(relativeName, "/") - if relativeName == "" { - continue - } - f := ManifestEntryToRemoteFile(entry, prefixInManifest) + for _, fileName := range fileNames { retry := retrier.New(retrier.ExponentialBackoff(RetriesOnFailure, common.AddRandomJitter(RetriesDuration, RetriesJitter)), RetrierClassifier) err := retry.RunCtx(ctx, func(ctx context.Context) error { - r, err := bd.GetFileReader(ctx, path.Join(remotePath, f.Name())) + r, err := bd.GetFileReader(ctx, path.Join(remotePath, fileName)) if err != nil { log.Error().Err(err).Send() return errors.WithMessage(err, "DownloadPathWithManifest GetFileReader") @@ -222,7 +400,7 @@ func (bd *BackupDestination) DownloadPathWithManifest(ctx context.Context, remot } }() defer close(watchDone) - dstFilePath := path.Join(localPath, f.Name()) + dstFilePath := path.Join(localPath, fileName) dstDirPath, _ := path.Split(dstFilePath) if err := os.MkdirAll(dstDirPath, 0750); err != nil { log.Error().Err(err).Send() diff --git a/pkg/storage/manifest_test.go b/pkg/storage/manifest_test.go index c94b3369..70accf9c 100644 --- a/pkg/storage/manifest_test.go +++ b/pkg/storage/manifest_test.go @@ -1,304 +1,212 @@ package storage import ( + "compress/gzip" + "fmt" + "io" + "os" + "sync" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func TestNewBackupManifest(t *testing.T) { - t.Parallel() - m := NewBackupManifest("test_backup") - assert.Equal(t, ManifestVersion, m.Version) - assert.Equal(t, "test_backup", m.BackupName) - assert.NotZero(t, m.CreatedAt) - assert.Empty(t, m.Files) - assert.Equal(t, 0, m.TotalFiles) - assert.Equal(t, int64(0), m.TotalSize) +// gunzipToTemp decompresses a .gz file into a temporary file and returns its path. +func gunzipToTemp(t *testing.T, gzPath string) string { + t.Helper() + src, err := os.Open(gzPath) + require.NoError(t, err) + defer src.Close() + gz, err := gzip.NewReader(src) + require.NoError(t, err) + dst, err := os.CreateTemp(t.TempDir(), "manifest-*.bolt") + require.NoError(t, err) + _, err = io.Copy(dst, gz) + require.NoError(t, err) + require.NoError(t, dst.Close()) + return dst.Name() } -func TestBackupManifest_AddFile(t *testing.T) { - t.Parallel() - m := NewBackupManifest("test_backup") - now := time.Now().UTC() - - m.AddFile("shadow/default/table1/default/part1/data.bin", 12345, now) - m.AddFile("shadow/default/table1/default/part1/primary.idx", 456, now) - m.AddFile("metadata/default/table1.json", 789, now) - - assert.Equal(t, 3, m.TotalFiles) - assert.Equal(t, int64(12345+456+789), m.TotalSize) - assert.Len(t, m.Files, 3) - - assert.Equal(t, "shadow/default/table1/default/part1/data.bin", m.Files[0].Path) - assert.Equal(t, int64(12345), m.Files[0].Size) - assert.Equal(t, now, m.Files[0].LastModified) +func finalizeAndOpen(t *testing.T, w *ManifestWriter) *ManifestReader { + t.Helper() + gzPath, err := w.Finalize() + require.NoError(t, err) + reader, err := openManifestReader(gunzipToTemp(t, gzPath)) + require.NoError(t, err) + return reader } -func TestBackupManifest_FilesUnderPrefix(t *testing.T) { +func TestManifestWriterReaderRoundtrip(t *testing.T) { t.Parallel() - m := NewBackupManifest("test_backup") - now := time.Now().UTC() - - // Add files from different tables and parts - m.AddFile("shadow/default/table1/default/part1/data.bin", 100, now) - m.AddFile("shadow/default/table1/default/part1/primary.idx", 50, now) - m.AddFile("shadow/default/table1/default/part2/data.bin", 200, now) - m.AddFile("shadow/default/table2/default/part1/data.bin", 300, now) - m.AddFile("metadata/default/table1.json", 10, now) - - // Files under part1 of table1 - files := m.FilesUnderPrefix("shadow/default/table1/default/part1") - assert.Len(t, files, 2) - for _, f := range files { - assert.Contains(t, f.Path, "shadow/default/table1/default/part1/") - } + w, err := NewManifestWriter("test_backup") + require.NoError(t, err) + defer w.Close() - // Files under table1 (all parts) - files = m.FilesUnderPrefix("shadow/default/table1") - assert.Len(t, files, 3) // part1/data.bin, part1/primary.idx, part2/data.bin + w.AddFile("shadow/default/table1/default/part1/data.bin") + w.AddFile("shadow/default/table1/default/part1/primary.idx") + w.AddFile("shadow/default/table1/default/part2/data.bin") + w.AddFile("shadow/default/table2/default/part1/data.bin") + w.AddFile("metadata/default/table1.json") - // Files under table2 - files = m.FilesUnderPrefix("shadow/default/table2") - assert.Len(t, files, 1) + assert.Equal(t, 5, w.TotalFiles) - // Files under metadata - files = m.FilesUnderPrefix("metadata") - assert.Len(t, files, 1) + reader := finalizeAndOpen(t, w) + defer reader.Close() - // Non-existent prefix - files = m.FilesUnderPrefix("shadow/default/table3") - assert.Len(t, files, 0) -} - -func TestBackupManifest_FilesUnderPrefix_TrailingSlash(t *testing.T) { - t.Parallel() - m := NewBackupManifest("test") - now := time.Now().UTC() - m.AddFile("shadow/db/tbl/disk/part1/file.bin", 100, now) - - // Should work with or without trailing slash - files1 := m.FilesUnderPrefix("shadow/db/tbl/disk/part1") - files2 := m.FilesUnderPrefix("shadow/db/tbl/disk/part1/") - assert.Equal(t, len(files1), len(files2)) -} - -func TestBackupManifest_MarshalUnmarshal(t *testing.T) { - t.Parallel() - m := NewBackupManifest("my_backup_2025") - now := time.Now().UTC().Truncate(time.Millisecond) // truncate for JSON roundtrip - - m.AddFile("shadow/default/orders/default/20250101_1_1_0/data.bin", 1048576, now) - m.AddFile("shadow/default/orders/default/20250101_1_1_0/primary.idx", 256, now) - m.AddFile("metadata/default/orders.json", 1024, now) + // Files under part1 of table1, names relative to the prefix + files, err := reader.FilesUnderPrefix("shadow/default/table1/default/part1") + require.NoError(t, err) + assert.Equal(t, []string{"data.bin", "primary.idx"}, files) - data, err := m.Marshal() + // Files under table1 (all parts) + files, err = reader.FilesUnderPrefix("shadow/default/table1") require.NoError(t, err) - assert.Contains(t, string(data), `"version": 1`) - assert.Contains(t, string(data), `"backup_name": "my_backup_2025"`) - assert.Contains(t, string(data), `"total_files": 3`) + assert.Equal(t, []string{"default/part1/data.bin", "default/part1/primary.idx", "default/part2/data.bin"}, files) - // Unmarshal - m2, err := UnmarshalManifest(data) + // Trailing slash is equivalent + files2, err := reader.FilesUnderPrefix("shadow/default/table1/") require.NoError(t, err) - assert.Equal(t, m.Version, m2.Version) - assert.Equal(t, m.BackupName, m2.BackupName) - assert.Equal(t, m.TotalFiles, m2.TotalFiles) - assert.Equal(t, m.TotalSize, m2.TotalSize) - assert.Len(t, m2.Files, 3) - - // Verify file entries round-trip - assert.Equal(t, m.Files[0].Path, m2.Files[0].Path) - assert.Equal(t, m.Files[0].Size, m2.Files[0].Size) - assert.Equal(t, m.Files[1].Path, m2.Files[1].Path) - assert.Equal(t, m.Files[2].Path, m2.Files[2].Path) -} + assert.Equal(t, files, files2) -func TestUnmarshalManifest_InvalidJSON(t *testing.T) { - t.Parallel() - _, err := UnmarshalManifest([]byte(`not json`)) - assert.Error(t, err) + // Non-existent prefix + files, err = reader.FilesUnderPrefix("shadow/default/table3") + require.NoError(t, err) + assert.Empty(t, files) } -func TestUnmarshalManifest_EmptyFiles(t *testing.T) { +func TestManifestReader_FilesUnderPrefix_NoPartialMatch(t *testing.T) { t.Parallel() - data := []byte(`{"version":1,"backup_name":"empty","created_at":"2025-05-16T00:00:00Z","total_size":0,"total_files":0,"files":[]}`) - m, err := UnmarshalManifest(data) + w, err := NewManifestWriter("test") require.NoError(t, err) - assert.Equal(t, "empty", m.BackupName) - assert.Empty(t, m.Files) -} + defer w.Close() -func TestManifestEntryToRemoteFile(t *testing.T) { - t.Parallel() - now := time.Now().UTC() - entry := ManifestEntry{ - Path: "shadow/default/table1/default/part1/data.bin", - Size: 12345, - LastModified: now, - } + // "part1" should not match "part10" or "part1_suffix" + w.AddFile("shadow/db/tbl/disk/part1/file.bin") + w.AddFile("shadow/db/tbl/disk/part10/file.bin") + w.AddFile("shadow/db/tbl/disk/part1_suffix/file.bin") - // No prefix stripping - rf := ManifestEntryToRemoteFile(entry, "") - assert.Equal(t, "shadow/default/table1/default/part1/data.bin", rf.Name()) - assert.Equal(t, int64(12345), rf.Size()) - assert.Equal(t, now, rf.LastModified()) + reader := finalizeAndOpen(t, w) + defer reader.Close() - // With prefix stripping - rf2 := ManifestEntryToRemoteFile(entry, "shadow/default/table1/default/part1") - assert.Equal(t, "data.bin", rf2.Name()) - assert.Equal(t, int64(12345), rf2.Size()) + files, err := reader.FilesUnderPrefix("shadow/db/tbl/disk/part1") + require.NoError(t, err) + assert.Equal(t, []string{"file.bin"}, files, "should only match exact prefix with trailing slash") } -func TestManifestEntryToRemoteFile_PrefixWithSlash(t *testing.T) { +func TestManifestWriter_BatchFlush(t *testing.T) { t.Parallel() - entry := ManifestEntry{ - Path: "shadow/db/tbl/disk/part/subdir/file.bin", - Size: 100, - } - rf := ManifestEntryToRemoteFile(entry, "shadow/db/tbl/disk/part/") - assert.Equal(t, "subdir/file.bin", rf.Name()) -} + w, err := NewManifestWriter("large_backup") + require.NoError(t, err) + defer w.Close() -func TestBackupManifest_LargeFileCount(t *testing.T) { - t.Parallel() - m := NewBackupManifest("large_backup") - now := time.Now().UTC() - - // Simulate a real backup with 10k files - for i := 0; i < 10000; i++ { - m.AddFile( - "shadow/default/trades/default/20250101_1_1_0/column"+string(rune('A'+i%26))+".bin", - int64(1024+i), - now, - ) + // Cross manifestBatchSize to force an intermediate flush plus the Finalize flush + total := manifestBatchSize + 100 + for i := 0; i < total; i++ { + w.AddFile(fmt.Sprintf("shadow/default/trades/default/part%05d/data.bin", i)) } + assert.Equal(t, total, w.TotalFiles) - assert.Equal(t, 10000, m.TotalFiles) - assert.Greater(t, m.TotalSize, int64(0)) - - // Marshal and unmarshal should work - data, err := m.Marshal() - require.NoError(t, err) + reader := finalizeAndOpen(t, w) + defer reader.Close() - m2, err := UnmarshalManifest(data) + files, err := reader.FilesUnderPrefix("shadow/default/trades") require.NoError(t, err) - assert.Equal(t, 10000, m2.TotalFiles) - assert.Equal(t, m.TotalSize, m2.TotalSize) + assert.Len(t, files, total) } -func TestManifestFileName_IsCorrect(t *testing.T) { +func TestManifestWriter_Concurrent(t *testing.T) { t.Parallel() - assert.Equal(t, "manifest.json", ManifestFileName) -} + w, err := NewManifestWriter("concurrent") + require.NoError(t, err) + defer w.Close() + + var wg sync.WaitGroup + for g := 0; g < 8; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + for i := 0; i < 500; i++ { + w.AddFile(fmt.Sprintf("shadow/db/tbl/disk/part%d/file%d.bin", g, i)) + } + }(g) + } + wg.Wait() + assert.Equal(t, 8*500, w.TotalFiles) -func TestManifestVersion_IsOne(t *testing.T) { - t.Parallel() - assert.Equal(t, 1, ManifestVersion) + reader := finalizeAndOpen(t, w) + defer reader.Close() + files, err := reader.FilesUnderPrefix("shadow/db/tbl/disk/part3") + require.NoError(t, err) + assert.Len(t, files, 500) } -func TestBackupManifest_HasFile(t *testing.T) { +func TestManifestWriter_EmptyManifest(t *testing.T) { t.Parallel() - m := NewBackupManifest("test") - now := time.Now().UTC() + w, err := NewManifestWriter("empty") + require.NoError(t, err) + defer w.Close() - m.AddFile("metadata/default/orders.json", 100, now) - m.AddFile("shadow/default/orders/default/part1/data.bin", 200, now) + reader := finalizeAndOpen(t, w) + defer reader.Close() - assert.True(t, m.HasFile("metadata/default/orders.json")) - assert.True(t, m.HasFile("shadow/default/orders/default/part1/data.bin")) - assert.False(t, m.HasFile("metadata/default/nonexistent.json")) - assert.False(t, m.HasFile("metadata/default/orders.json/")) - assert.False(t, m.HasFile("")) + files, err := reader.FilesUnderPrefix("shadow/db/tbl") + require.NoError(t, err) + assert.Empty(t, files) } -func TestNewBackupManifestWithCapacity(t *testing.T) { +func TestManifestWriter_CloseRemovesTempFiles(t *testing.T) { t.Parallel() - m := NewBackupManifestWithCapacity("test_backup", 5000) - assert.Equal(t, ManifestVersion, m.Version) - assert.Equal(t, "test_backup", m.BackupName) - assert.NotZero(t, m.CreatedAt) - assert.Empty(t, m.Files) - assert.Equal(t, 5000, cap(m.Files)) - assert.Equal(t, 0, m.TotalFiles) - assert.Equal(t, int64(0), m.TotalSize) -} + w, err := NewManifestWriter("cleanup") + require.NoError(t, err) + w.AddFile("shadow/db/tbl/disk/part/file.bin") -func TestNewBackupManifestWithCapacity_MinimumFloor(t *testing.T) { - t.Parallel() - m := NewBackupManifestWithCapacity("test", 10) - assert.Equal(t, 256, cap(m.Files), "capacity should be at least 256") -} + gzPath, err := w.Finalize() + require.NoError(t, err) + boltPath := w.localPath + _, err = os.Stat(gzPath) + require.NoError(t, err) -func TestNewBackupManifestWithCapacity_ZeroCapacity(t *testing.T) { - t.Parallel() - m := NewBackupManifestWithCapacity("test", 0) - assert.Equal(t, 256, cap(m.Files), "zero capacity should use floor of 256") + w.Close() + _, err = os.Stat(gzPath) + assert.True(t, os.IsNotExist(err)) + _, err = os.Stat(boltPath) + assert.True(t, os.IsNotExist(err)) + // Close is idempotent and nil-safe + w.Close() + (*ManifestWriter)(nil).Close() } -func TestNewBackupManifestWithCapacity_NegativeCapacity(t *testing.T) { +func TestManifestWriter_FinalizeTwiceFails(t *testing.T) { t.Parallel() - m := NewBackupManifestWithCapacity("test", -100) - assert.Equal(t, 256, cap(m.Files), "negative capacity should use floor of 256") + w, err := NewManifestWriter("twice") + require.NoError(t, err) + defer w.Close() + + _, err = w.Finalize() + require.NoError(t, err) + _, err = w.Finalize() + assert.Error(t, err) } -func TestNewBackupManifestWithCapacity_CorrectBehavior(t *testing.T) { +func TestOpenManifestReader_InvalidFile(t *testing.T) { t.Parallel() - // Pre-sized manifest should produce identical results to default manifest - now := time.Now().UTC() - m1 := NewBackupManifest("test") - m2 := NewBackupManifestWithCapacity("test", 10000) - - for i := 0; i < 1000; i++ { - path := "shadow/default/table/default/part/col" + string(rune('A'+i%26)) + ".bin" - m1.AddFile(path, int64(i*100), now) - m2.AddFile(path, int64(i*100), now) - } + tmp, err := os.CreateTemp(t.TempDir(), "garbage-*.bolt") + require.NoError(t, err) + _, err = tmp.WriteString("not a bolt database") + require.NoError(t, err) + require.NoError(t, tmp.Close()) - assert.Equal(t, m1.TotalFiles, m2.TotalFiles) - assert.Equal(t, m1.TotalSize, m2.TotalSize) - assert.Equal(t, len(m1.Files), len(m2.Files)) - - // Verify marshal/unmarshal produces identical output - data1, err1 := m1.Marshal() - data2, err2 := m2.Marshal() - require.NoError(t, err1) - require.NoError(t, err2) - - // Unmarshal and compare (can't compare JSON directly due to timestamp) - um1, _ := UnmarshalManifest(data1) - um2, _ := UnmarshalManifest(data2) - assert.Equal(t, um1.TotalFiles, um2.TotalFiles) - assert.Equal(t, um1.TotalSize, um2.TotalSize) + _, err = openManifestReader(tmp.Name()) + assert.Error(t, err) } -func TestNewBackupManifestWithCapacity_LargeCapacityNoAlloc(t *testing.T) { +func TestManifestFileName_IsCorrect(t *testing.T) { t.Parallel() - // Verify that pre-sizing to exact capacity produces correct results - m := NewBackupManifestWithCapacity("test", 50000) - now := time.Now().UTC() - - for i := 0; i < 50000; i++ { - m.AddFile("shadow/default/t/d/p/f.bin", int64(i), now) - } - assert.Equal(t, 50000, m.TotalFiles) - // Pre-sized capacity should prevent any reallocation - assert.GreaterOrEqual(t, cap(m.Files), 50000) + assert.Equal(t, "manifest.bolt.gz", ManifestFileName) } -func TestBackupManifest_FilesUnderPrefix_NoPartialMatch(t *testing.T) { +func TestManifestVersion_IsOne(t *testing.T) { t.Parallel() - m := NewBackupManifest("test") - now := time.Now().UTC() - - // "part1" should not match "part10" or "part1_suffix" - m.AddFile("shadow/db/tbl/disk/part1/file.bin", 100, now) - m.AddFile("shadow/db/tbl/disk/part10/file.bin", 200, now) - m.AddFile("shadow/db/tbl/disk/part1_suffix/file.bin", 300, now) - - files := m.FilesUnderPrefix("shadow/db/tbl/disk/part1") - assert.Len(t, files, 1, "should only match exact prefix with trailing slash") - assert.Equal(t, "shadow/db/tbl/disk/part1/file.bin", files[0].Path) + assert.Equal(t, 1, ManifestVersion) } From 9892ecbc0de5419d93745877f773567240478eb8 Mon Sep 17 00:00:00 2001 From: slach Date: Thu, 9 Jul 2026 14:37:36 +0400 Subject: [PATCH 4/5] fix TestRebase race condition and properly cleanup --- test/integration/rebase_test.go | 58 +++++++++++++++++++++++---------- 1 file changed, 40 insertions(+), 18 deletions(-) diff --git a/test/integration/rebase_test.go b/test/integration/rebase_test.go index 28f750af..11038a7f 100644 --- a/test/integration/rebase_test.go +++ b/test/integration/rebase_test.go @@ -150,11 +150,19 @@ func runRebaseScenario(t *testing.T, r *require.Assertions, env *TestEnvironment inc1Backup := dbName + "_inc1" inc2Backup := dbName + "_inc2" - for _, b := range []string{fullBackup, inc1Backup, inc2Backup} { - env.DockerExecNoError(r, "clickhouse-backup", "bash", "-ce", "clickhouse-backup -c "+configFile+" delete remote "+b+" 2>/dev/null || true") - env.DockerExecNoError(r, "clickhouse-backup", "bash", "-ce", "clickhouse-backup -c "+configFile+" delete local "+b+" 2>/dev/null || true") + // deferred (not t.Cleanup) so it unwinds before the caller's `defer env.Cleanup(t, r)`, + // while the ClickHouse connection is still open: a mid-scenario `require` failure calls + // t.FailNow()/runtime.Goexit and skips the rest of this function, so without this the + // shared pooled env would leak test_rebase_* state into whichever test reuses it next + cleanupScenario := func() { + for _, b := range []string{fullBackup, inc1Backup, inc2Backup} { + env.DockerExecNoError(r, "clickhouse-backup", "bash", "-ce", "clickhouse-backup -c "+configFile+" delete remote "+b+" 2>/dev/null || true") + env.DockerExecNoError(r, "clickhouse-backup", "bash", "-ce", "clickhouse-backup -c "+configFile+" delete local "+b+" 2>/dev/null || true") + } + r.NoError(env.dropDatabase(dbName, true)) } - r.NoError(env.dropDatabase(dbName, true)) + defer cleanupScenario() + cleanupScenario() env.queryWithNoError(t, r, "CREATE DATABASE "+dbName) for _, table := range tables { @@ -201,20 +209,39 @@ func runRebaseScenario(t *testing.T, r *require.Assertions, env *TestEnvironment env.DockerExecNoError(r, "clickhouse-backup", "clickhouse-backup", "-c", configFile, "rebase", inc2Backup) // the rebased backup on remote storage must keep the chain data_format and lose `required_backup`, - // check via `list remote` (local metadata.json is unusable here - download clears `data_format`) - out, err = env.DockerExecOut("clickhouse-backup", "clickhouse-backup", "-c", configFile, "list", "remote") - r.NoError(err, "list remote: %s", out) - inc2Line := "" + // check via `list remote --format=json` (local metadata.json is unusable here - download clears + // `data_format`; json avoids scraping the human `text` table, whose tabwriter output can be spliced + // mid-row by a concurrent zerolog write to the same stdout/stderr stream) + out, err = env.DockerExecOut("clickhouse-backup", "clickhouse-backup", "-c", configFile, "list", "remote", "--format=json") + r.NoError(err, "list remote --format=json: %s", out) + // `out` also contains clickhouse-backup's own log lines around the JSON payload, the JSON array + // itself is written as a single line via fmt.Fprintln, so pick that line out before unmarshalling + jsonLine := "" for _, line := range strings.Split(out, "\n") { - if strings.HasPrefix(line, inc2Backup+" ") { - inc2Line = line + if strings.HasPrefix(strings.TrimSpace(line), "[") { + jsonLine = line + break + } + } + r.NotEmptyf(jsonLine, "no JSON array found in `list remote --format=json` output: %s", out) + type remoteBackupInfo struct { + BackupName string `json:"BackupName"` + RequiredBackup string `json:"RequiredBackup"` + Description string `json:"Description"` + } + var remoteBackups []remoteBackupInfo + r.NoError(json.Unmarshal([]byte(jsonLine), &remoteBackups), "list remote --format=json: %s", out) + var inc2Info *remoteBackupInfo + for i := range remoteBackups { + if remoteBackups[i].BackupName == inc2Backup { + inc2Info = &remoteBackups[i] break } } - r.NotEmptyf(inc2Line, "%s not found in `list remote` output: %s", inc2Backup, out) - r.NotContainsf(inc2Line, "+", "expected empty required_backup after rebase") + r.NotNilf(inc2Info, "%s not found in `list remote --format=json` output: %s", inc2Backup, out) + r.Emptyf(inc2Info.RequiredBackup, "expected empty required_backup after rebase") if expectedDataFormat != "" { - r.Containsf(inc2Line, expectedDataFormat, "expected data_format=%s after rebase", expectedDataFormat) + r.Containsf(inc2Info.Description, expectedDataFormat, "expected data_format=%s after rebase", expectedDataFormat) } // ancestors are not needed anymore @@ -258,9 +285,4 @@ func runRebaseScenario(t *testing.T, r *require.Assertions, env *TestEnvironment } r.Equalf(3, partsCount, "%s: expected 3 parts after rebase; metadata=%s", table, out) } - - // Cleanup - env.DockerExecNoError(r, "clickhouse-backup", "clickhouse-backup", "-c", configFile, "delete", "remote", inc2Backup) - env.DockerExecNoError(r, "clickhouse-backup", "clickhouse-backup", "-c", configFile, "delete", "local", inc2Backup) - r.NoError(env.dropDatabase(dbName, true)) } From c00da1390f44c8de585e4763756e5ed632dcd6c3 Mon Sep 17 00:00:00 2001 From: slach Date: Thu, 9 Jul 2026 15:08:38 +0400 Subject: [PATCH 5/5] update go.mod --- go.mod | 26 +++++++++++++------------- go.sum | 27 +++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 13 deletions(-) diff --git a/go.mod b/go.mod index 8d2e9421..ad3b79ae 100644 --- a/go.mod +++ b/go.mod @@ -8,9 +8,9 @@ require ( github.com/ClickHouse/clickhouse-go/v2 v2.47.0 github.com/antchfx/xmlquery v1.5.1 github.com/aws/aws-sdk-go-v2 v1.42.1 - github.com/aws/aws-sdk-go-v2/config v1.32.28 - github.com/aws/aws-sdk-go-v2/credentials v1.19.27 - github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.14 + github.com/aws/aws-sdk-go-v2/config v1.32.29 + github.com/aws/aws-sdk-go-v2/credentials v1.19.28 + github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.3.1 github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0 github.com/aws/aws-sdk-go-v2/service/sts v1.44.0 github.com/aws/smithy-go v1.27.3 @@ -47,18 +47,18 @@ require ( github.com/xyproto/gionice v1.3.0 github.com/yargevad/filepathx v1.0.0 go.etcd.io/bbolt v1.5.0 - golang.org/x/crypto v0.53.0 - golang.org/x/mod v0.37.0 - golang.org/x/sync v0.21.0 - golang.org/x/text v0.39.0 - google.golang.org/api v0.287.0 + golang.org/x/crypto v0.54.0 + golang.org/x/mod v0.38.0 + golang.org/x/sync v0.22.0 + golang.org/x/text v0.40.0 + google.golang.org/api v0.287.1 gopkg.in/yaml.v3 v3.0.1 ) require ( cel.dev/expr v0.25.2 // indirect cloud.google.com/go v0.123.0 // indirect - cloud.google.com/go/auth v0.20.0 // indirect + cloud.google.com/go/auth v0.21.0 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect cloud.google.com/go/iam v1.11.0 // indirect @@ -82,7 +82,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.30 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31 // indirect - github.com/aws/aws-sdk-go-v2/service/signin v1.3.0 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.4.0 // indirect github.com/aws/aws-sdk-go-v2/service/sso v1.32.0 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.37.0 // indirect github.com/beorn7/perks v1.0.1 // indirect @@ -112,7 +112,7 @@ require ( github.com/google/go-querystring v1.2.0 // indirect github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.18 // indirect - github.com/googleapis/gax-go/v2 v2.22.0 // indirect + github.com/googleapis/gax-go/v2 v2.23.0 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/kr/fs v0.1.0 // indirect github.com/kylelemons/godebug v1.1.0 // indirect @@ -163,9 +163,9 @@ require ( go.opentelemetry.io/otel/trace v1.44.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect go4.org v0.0.0-20260112195520-a5071408f32f // indirect - golang.org/x/net v0.56.0 // indirect + golang.org/x/net v0.57.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sys v0.46.0 // indirect + golang.org/x/sys v0.47.0 // indirect golang.org/x/time v0.15.0 // indirect google.golang.org/genproto v0.0.0-20260706201446-f0a921348800 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260706201446-f0a921348800 // indirect diff --git a/go.sum b/go.sum index 20f511e1..e66b814d 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= cloud.google.com/go/auth v0.20.0 h1:kXTssoVb4azsVDoUiF8KvxAqrsQcQtB53DcSgta74CA= cloud.google.com/go/auth v0.20.0/go.mod h1:942/yi/itH1SsmpyrbnTMDgGfdy2BUqIKyd0cyYLc5Q= +cloud.google.com/go/auth v0.21.0 h1:g/QwYfYb2Ai6HH8oomAOyBaIHLbscZ4+T/F/f5JZHkE= +cloud.google.com/go/auth v0.21.0/go.mod h1:M9o2Oz+YI2jAfxewJgb1vyI3vceHF+eohmxyzmrl+9s= cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute v1.64.0 h1:7MmuzeAxlG5MOG5PQD2NLtyYR6bWjkvGljRu7pByoRU= @@ -93,12 +95,16 @@ github.com/aws/aws-sdk-go-v2/config v1.32.26 h1:JI+W5B3jUA8UBz2ggbICGd9UCR6/+SB2 github.com/aws/aws-sdk-go-v2/config v1.32.26/go.mod h1:RLE2Ls/wRstvdSz1GPrIWNnXcKZ/znDdWyMuiQxdBoY= github.com/aws/aws-sdk-go-v2/config v1.32.28 h1:qY6afygxK5c2PPU3Sz8W6yB5W44RF1vnmPdBwViDN+Y= github.com/aws/aws-sdk-go-v2/config v1.32.28/go.mod h1:WeS/wN1IDs8YC+BxTrFz9ZyJ1rufRBQfirOcDusEpmQ= +github.com/aws/aws-sdk-go-v2/config v1.32.29 h1:BcMHHnpiWKogf+gGfpj3K1w+Sktz29XDo/cPSAPO3FU= +github.com/aws/aws-sdk-go-v2/config v1.32.29/go.mod h1:+Kbhn8Es4kPUph3F/0W7avykytc+Jh2Ld9/msv9ljV4= github.com/aws/aws-sdk-go-v2/credentials v1.19.22 h1:SHfH6wyPsEgG7fVsi5rQxWEt7tuIcN2PGhb1mTFv6tE= github.com/aws/aws-sdk-go-v2/credentials v1.19.22/go.mod h1:54nO8lKD4aQPOntM/VTWjnR+DYzTwx0YkSMZMhAgewQ= github.com/aws/aws-sdk-go-v2/credentials v1.19.25 h1:TzPVjfUZ1hsKafvYE+DIzKXIik2KufQxsPHanlkttbo= github.com/aws/aws-sdk-go-v2/credentials v1.19.25/go.mod h1:K4hw0buguVvtC74HnVfTRr0LzQQHAWPqJbBU9QGk2Pg= github.com/aws/aws-sdk-go-v2/credentials v1.19.27 h1:cFksKkdaBGGmpe6XJpvrxFNWkbXY5/gwFqZNB2O9WCM= github.com/aws/aws-sdk-go-v2/credentials v1.19.27/go.mod h1:20CoObBgNhFfl8/ggDQu2IZmItxDhkLcWSy4C3alDPI= +github.com/aws/aws-sdk-go-v2/credentials v1.19.28 h1:zTXJSsNcoO91/mTXsZoYf0AK8dvNPiA58/VtyGXR+wM= +github.com/aws/aws-sdk-go-v2/credentials v1.19.28/go.mod h1:Kd9E0JzDBW/q1xbsHFrev/GnbAf5J0Ng8xoyc7HZ91Q= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28 h1:b+kcDejJrXc30zU/w8Tc9klISwaO5wh+6T0sMBdDoHM= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28/go.mod h1:LnI62O9GnSv6GcuLXxOYqlq0C8EmxMcgnF6m7LdYuOY= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.29 h1:r6qZHbT+wxgWO/e9vYNUEtg7lv5+UN3pRqKhLXvnArg= @@ -111,6 +117,8 @@ github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.12 h1:0jSSxdkgDH6Sh github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.12/go.mod h1:rdg/WbW8wInpiUA/6qzLDwn97IPD8d3pvfMbCBWslMA= github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.14 h1:oBzlDywQsCbpL0+GRlS5ycXQkR7lmuzeHcLl0ycAQ0Q= github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.14/go.mod h1:ED1gaDLOD0VEAEQiJgobHSM/iFRTeH++5wjV3IwvoQ4= +github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.3.1 h1:in1rmZHNBr3OntzYxPRKnjW7eCYLrTExI0VQXR+C8AY= +github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.3.1/go.mod h1:KYWha/ENCwaAgWraz9liTvWPx59WRs3yJDSu1cjk5io= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28 h1:Xf2j7NdVcUKomlZ4iihOP4AZ3Fzlr8h4yKpXeP+OFPg= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28/go.mod h1:O8cDo1dW63jU7ki//kRe1z+tLGcpnD1jrouitsQddDw= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.29 h1:f3vKqSo13fhTYb+JEcXwXefZQE26I1FB5eTSniU67ko= @@ -163,6 +171,8 @@ github.com/aws/aws-sdk-go-v2/service/signin v1.2.1 h1:BeJmkm5YOZs6lGRGcNoIuLSoTT github.com/aws/aws-sdk-go-v2/service/signin v1.2.1/go.mod h1:LxYujSTLPRlp2vTtcUO/+1ilrew8ytt6SvQyOgejzFQ= github.com/aws/aws-sdk-go-v2/service/signin v1.3.0 h1:i0+tbB9QBnzL5NrF2WR/zk8q2s+1N+RaDYr2627E8UI= github.com/aws/aws-sdk-go-v2/service/signin v1.3.0/go.mod h1:mxC0nT/C8wMMS97DemZPzvUZxvIt+2Iq+eS3JdFZGgg= +github.com/aws/aws-sdk-go-v2/service/signin v1.4.0 h1:sLzmJGCMv+C8KqiJgEqDLB6vxaJGmobRh4rr//ZpA3w= +github.com/aws/aws-sdk-go-v2/service/signin v1.4.0/go.mod h1:mxC0nT/C8wMMS97DemZPzvUZxvIt+2Iq+eS3JdFZGgg= github.com/aws/aws-sdk-go-v2/service/sso v1.31.2 h1:ySNWu7TPmj5fKFIa1GYvX+Ddxd5ccruqC20aMNuyWDM= github.com/aws/aws-sdk-go-v2/service/sso v1.31.2/go.mod h1:A+U9luAOwFeB1kseyWCITVg7/NntoPebCFR9pQ4ch9A= github.com/aws/aws-sdk-go-v2/service/sso v1.31.4 h1:i465b/3c7xJd++pobNIDOggouekCuiWOnB0goQJy+94= @@ -299,6 +309,8 @@ github.com/googleapis/enterprise-certificate-proxy v0.3.18 h1:hvVi34VucdrV1IIsiW github.com/googleapis/enterprise-certificate-proxy v0.3.18/go.mod h1:rSEsBUemEBZEexP2y6jPp16LUmUbjmSbcPMQizR0o4k= github.com/googleapis/gax-go/v2 v2.22.0 h1:PjIWBpgGIVKGoCXuiCoP64altEJCj3/Ei+kSU5vlZD4= github.com/googleapis/gax-go/v2 v2.22.0/go.mod h1:irWBbALSr0Sk3qlqb9SyJ1h68WjgeFuiOzI4Rqw5+aY= +github.com/googleapis/gax-go/v2 v2.23.0 h1:Tchl7qkvE7Ip3y+ztvNufYFvkfqTe7NfLTYGIdJRLuE= +github.com/googleapis/gax-go/v2 v2.23.0/go.mod h1:rBQKOVJCdb8IFEzg+FCwlt1LP/xMDGuqUXhUG+XMXEg= github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= @@ -536,6 +548,8 @@ golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= +golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= @@ -545,6 +559,8 @@ golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4= golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= +golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= +golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= @@ -558,6 +574,8 @@ golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= +golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -569,6 +587,8 @@ golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -587,6 +607,8 @@ golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= @@ -599,6 +621,7 @@ golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM= golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/term v0.44.0 h1:0rLvDRCtNj0gZkyIXhCyOb2OAzEhLVqc4B+hrsBhrmc= +golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= @@ -614,6 +637,8 @@ golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -632,6 +657,8 @@ google.golang.org/api v0.286.0 h1:TdTXMvzYKnWV1/lPbCdbXRqBrkDqjPto22H2xeZZ8LI= google.golang.org/api v0.286.0/go.mod h1:NlOlUIr8MPoIhT9Bb/oUnRuHbJOLwxb6JSYJM8Yz+jQ= google.golang.org/api v0.287.0 h1:CQDMqUiqZZ0U/Yge3zyjAhNQ0OSYEH0PaA7l4xtEen4= google.golang.org/api v0.287.0/go.mod h1:pPW85yt3Iuc3unkpaMhFtMmOqnTdCwCqEOaUlnuxRlQ= +google.golang.org/api v0.287.1 h1:LiyJx32VU3cwQfLchn/513qKhc25hq0pEANYJoWNnnI= +google.golang.org/api v0.287.1/go.mod h1:lM2kYRzYUCBY91P9h6VF1PYmvhxii3O5hji37qRvIcY= google.golang.org/genproto v0.0.0-20260526163538-3dc84a4a5aaa h1:mfj8IS4EA4VAR9a6QDVxTQkLY64iBybb5QI1B4pXrpE= google.golang.org/genproto v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:fuT7yonGw1Iq2oa+YC0fyqPPQJkgo/54gPNC6VitOkI= google.golang.org/genproto v0.0.0-20260622175928-b703f567277d h1:CP5omUq8AJTiWMrPKM1WRLJ7zZeXd9OPcQD3TbBNAyY=