diff --git a/docs-site/docs/features/backup-types.md b/docs-site/docs/features/backup-types.md index 5285c20..f41b946 100644 --- a/docs-site/docs/features/backup-types.md +++ b/docs-site/docs/features/backup-types.md @@ -12,12 +12,22 @@ When a task is routed to a remote Agent, the source tools and paths are resolved ## File / Directory -Tars (and optionally gzips) one or more filesystem paths. +File tasks offer three backup modes: + +- **Full archive** — writes a self-contained tar artifact on every run +- **Differential archive** — writes only changes since the current full baseline and periodically refreshes that baseline +- **CDC repository** — splits content with stable 512 KiB / 1 MiB / 4 MiB boundaries, stores new chunks in immutable 32 MiB packs, and writes a small snapshot manifest for each run + +The CDC repository deduplicates identical content across files and snapshots. Restore, selective restore, verification, download-as-tar, retention, and garbage collection all resolve data through the repository index. Compression and encryption are applied per chunk; encrypted repositories use keyed chunk IDs so plaintext hashes are not exposed. + +Repository mode currently uses a single-writer index and therefore runs on the Master only. To keep repository copies on multiple backends, select multiple primary storage targets on the task. Object-level replication is intentionally disabled because a snapshot manifest without its shared packs and indexes is not a complete backup. + +Common file-task options: - **Source** accepts multiple paths — one per line in the UI - **Exclude patterns** accept gitignore-style globs - Supports following symlinks, preserving permissions -- Output is a single `.tar` or `.tar.gz` artifact +- Full and differential modes output `.tar`, `.tar.gz`, or `.tar.zst` artifacts ## MySQL diff --git a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/backup-types.md b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/backup-types.md index df97237..bd6042e 100644 --- a/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/backup-types.md +++ b/docs-site/i18n/zh-CN/docusaurus-plugin-content-docs/current/features/backup-types.md @@ -12,12 +12,22 @@ BackupX 支持五种内置备份类型，类型决定了用哪个 runner 执行 ## 文件 / 目录 -打包（可选 gzip）一个或多个文件系统路径。 +文件任务提供三种备份模式： + +- **全量归档** — 每次生成一份可独立使用的 tar 产物 +- **差异归档** — 只保存相对当前全量基线的变化，并按周期刷新全量基线 +- **CDC 去重仓库** — 按稳定的 512 KiB / 1 MiB / 4 MiB 内容边界切块，将新块合并到不可变的 32 MiB pack，每次运行只新增一份小型快照清单 + +CDC 仓库会在不同文件、不同快照之间复用相同内容。完整恢复、选择性恢复、完整性校验、下载为 tar、保留策略和垃圾回收都通过仓库索引定位分块。压缩与加密按块执行；启用加密时使用带密钥的块 ID，不暴露明文哈希。 + +当前仓库索引采用单写者模型，因此 CDC 模式仅在 Master 本机执行。如需保存多份完整仓库，请在任务中直接多选主存储目标。对象级副本复制会被禁用，因为只有快照清单、没有共享 pack 与索引并不是完整备份。 + +文件任务的通用选项： - **源路径** 支持多个（UI 中每行一个） - **排除模式** 支持 gitignore 风格的通配符 - 可选跟随符号链接、保留权限 -- 输出单个 `.tar` 或 `.tar.gz` +- 全量与差异模式输出 `.tar`、`.tar.gz` 或 `.tar.zst` ## MySQL diff --git a/server/internal/app/app.go b/server/internal/app/app.go index 88deb5f..281b3e9 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -84,7 +84,7 @@ func New(ctx context.Context, cfg config.Config, version string) (*Application, // nodeRepo 在下方 Cluster 节点管理区块才实例化，这里延后注入 backupRunnerRegistry := backup.NewRegistry(backup.NewFileRunner(), backup.NewSQLiteRunner(), backup.NewMySQLRunner(nil), backup.NewPostgreSQLRunner(nil), backup.NewSAPHANARunner(nil), backup.NewMongoDBRunner(nil)) logHub := backup.NewLogHub() - retentionService := backupretention.NewService(backupRecordRepo) + retentionService := backupretention.NewService(backupRecordRepo, configCipher.Key()) notifyRegistry := notify.NewRegistry(notify.NewEmailNotifier(), notify.NewWebhookNotifier(), notify.NewTelegramNotifier()) notificationService := service.NewNotificationService(notificationRepo, notifyRegistry, configCipher) authService.SetNotificationService(notificationService) diff --git a/server/internal/backup/repository.go b/server/internal/backup/repository.go new file mode 100644 index 0000000..54b3f49 --- /dev/null +++ b/server/internal/backup/repository.go @@ -0,0 +1,1211 @@ +package backup + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "context" + "crypto/aes" + "crypto/cipher" + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path" + "path/filepath" + "sort" + "strings" + "time" + + "backupx/server/internal/storage" + "github.com/klauspost/compress/zstd" +) + +const ( + repositoryFormatVersion = 1 + repositoryRoot = ".backupx/repository/v1" + repositoryIndexPrefix = repositoryRoot + "/indexes" + repositoryPackPrefix = repositoryRoot + "/packs" + repositorySnapshotRoot = repositoryRoot + "/snapshots" + repositoryDefaultPack = int64(32 << 20) +) + +type RepositoryStore struct { + key []byte + chunker *contentDefinedChunker + packSize int64 +} + +type RepositoryPlan struct { + tempDir string + spoolPath string + chunks map[string]repositoryPlanChunk + chunkOrder []string + snapshot repositorySnapshot + Manifest Manifest + LogicalSize int64 + UniqueSize int64 +} + +type repositoryPlanChunk struct { + Offset int64 + Size int64 +} + +type repositorySnapshot struct { + Version int `json:"version"` + TaskID uint `json:"taskId"` + CreatedAt time.Time `json:"createdAt"` + Compression string `json:"compression"` + Encrypted bool `json:"encrypted"` + SourcePaths []string `json:"sourcePaths"` + Entries []repositoryEntry `json:"entries"` +} + +type repositoryEntry struct { + Path string `json:"path"` + Kind string `json:"kind"` + Mode uint32 `json:"mode"` + ModTime int64 `json:"modTime"` + Size int64 `json:"size"` + LinkTarget string `json:"linkTarget,omitempty"` + Chunks []string `json:"chunks,omitempty"` +} + +type repositorySnapshotEnvelope struct { + Version int `json:"version"` + Encrypted bool `json:"encrypted"` + Data json.RawMessage `json:"data,omitempty"` + Ciphertext string `json:"ciphertext,omitempty"` +} + +type repositoryChunkLocation struct { + Pack string `json:"pack"` + Offset int64 `json:"offset"` + Length int64 `json:"length"` + PlainSize int64 `json:"plainSize"` + Compression string `json:"compression"` + Encrypted bool `json:"encrypted"` +} + +type repositoryIndexSegment struct { + Version int `json:"version"` + CreatedAt time.Time `json:"createdAt"` + Pack string `json:"pack"` + Chunks map[string]repositoryChunkLocation `json:"chunks"` +} + +type RepositoryUploadResult struct { + SnapshotKey string + SnapshotSize int64 + LogicalSize int64 + UploadedBytes int64 + ReusedBytes int64 + UniqueChunks int + NewChunks int + Checksum string +} + +type RepositoryVerifyResult struct { + Entries int + Chunks int + Bytes int64 +} + +type RepositoryPruneResult struct { + DeletedIndexes int + DeletedPacks int + ReclaimedBytes int64 +} + +func NewRepositoryStore(encryptionKey []byte) *RepositoryStore { + keyCopy := make([]byte, len(encryptionKey)) + copy(keyCopy, encryptionKey) + return &RepositoryStore{ + key: keyCopy, + chunker: newContentDefinedChunker(), + packSize: repositoryDefaultPack, + } +} + +func (s *RepositoryStore) SnapshotKey(taskID, recordID uint, startedAt time.Time) string { + stamp := startedAt.UTC().Format("20060102T150405.000000000Z") + return fmt.Sprintf("%s/%d/%s-%d.bxrs", repositorySnapshotRoot, taskID, stamp, recordID) +} + +func (p *RepositoryPlan) Close() error { + if p == nil || strings.TrimSpace(p.tempDir) == "" { + return nil + } + err := os.RemoveAll(p.tempDir) + p.tempDir = "" + p.spoolPath = "" + return err +} + +func (s *RepositoryStore) BuildPlan(ctx context.Context, task TaskSpec, writer LogWriter) (plan *RepositoryPlan, err error) { + if writer == nil { + writer = NopLogWriter{} + } + sourcePaths := compactPaths(task.SourcePaths) + if len(sourcePaths) == 0 && strings.TrimSpace(task.SourcePath) != "" { + sourcePaths = []string{filepath.Clean(strings.TrimSpace(task.SourcePath))} + } + if len(sourcePaths) == 0 { + return nil, fmt.Errorf("source path is required") + } + compression, err := s.normalizeCompression(task.Compression) + if err != nil { + return nil, err + } + if task.Encrypt && len(s.key) != 32 { + return nil, fmt.Errorf("repository encryption requires a 256-bit key") + } + if err := os.MkdirAll(task.TempDir, 0o755); err != nil { + return nil, fmt.Errorf("create repository temp root: %w", err) + } + tempDir, err := os.MkdirTemp(task.TempDir, "repository-plan-*") + if err != nil { + return nil, fmt.Errorf("create repository plan directory: %w", err) + } + defer func() { + if err == nil { + return + } + if cleanupErr := os.RemoveAll(tempDir); cleanupErr != nil { + err = errors.Join(err, fmt.Errorf("clean repository plan: %w", cleanupErr)) + } + }() + + spoolPath := filepath.Join(tempDir, "chunks.spool") + spool, err := os.Create(spoolPath) + if err != nil { + return nil, fmt.Errorf("create repository chunk spool: %w", err) + } + spoolClosed := false + defer func() { + if spoolClosed { + return + } + if closeErr := spool.Close(); closeErr != nil { + err = errors.Join(err, fmt.Errorf("close repository chunk spool: %w", closeErr)) + } + }() + + plan = &RepositoryPlan{ + tempDir: tempDir, + spoolPath: spoolPath, + chunks: make(map[string]repositoryPlanChunk), + snapshot: repositorySnapshot{ + Version: repositoryFormatVersion, + TaskID: task.ID, + CreatedAt: task.StartedAt.UTC(), + Compression: compression, + Encrypted: task.Encrypt, + SourcePaths: sourcePaths, + Entries: make([]repositoryEntry, 0), + }, + Manifest: Manifest{Entries: make([]ManifestEntry, 0)}, + } + excludes := normalizeExcludePatterns(task.ExcludePatterns) + seenPaths := make(map[string]struct{}) + spoolOffset := int64(0) + writer.WriteLine(fmt.Sprintf("CDC 仓库模式：FastCDC %d KiB/%d KiB/%d KiB，pack %d MiB", repositoryChunkMin>>10, repositoryChunkAvg>>10, repositoryChunkMax>>10, s.packSize>>20)) + + for sourceIndex, rawSource := range sourcePaths { + sourcePath := filepath.Clean(strings.TrimSpace(rawSource)) + if _, statErr := os.Lstat(sourcePath); statErr != nil { + return nil, fmt.Errorf("stat source path %s: %w", sourcePath, statErr) + } + baseParent := filepath.Dir(sourcePath) + writer.WriteLine(fmt.Sprintf("扫描源路径 [%d/%d]：%s", sourceIndex+1, len(sourcePaths), sourcePath)) + walkErr := filepath.Walk(sourcePath, func(currentPath string, info os.FileInfo, walkErr error) error { + if walkErr != nil { + return walkErr + } + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr + } + relative, relErr := filepath.Rel(baseParent, currentPath) + if relErr != nil { + return relErr + } + archiveName := path.Clean(filepath.ToSlash(relative)) + if archiveName == "." || strings.HasPrefix(archiveName, "../") { + return fmt.Errorf("invalid repository entry path %q", archiveName) + } + if shouldExcludeEntry(archiveName, info.IsDir(), excludes) { + if info.IsDir() { + return filepath.SkipDir + } + return nil + } + if _, duplicated := seenPaths[archiveName]; duplicated { + return fmt.Errorf("duplicate repository entry %q from overlapping source paths", archiveName) + } + seenPaths[archiveName] = struct{}{} + + entry := repositoryEntry{ + Path: archiveName, + Mode: uint32(info.Mode().Perm()), + ModTime: info.ModTime().UTC().UnixNano(), + Size: info.Size(), + } + manifestEntry := entryFromInfo(archiveName, info) + switch { + case info.IsDir(): + entry.Kind = "directory" + case info.Mode()&os.ModeSymlink != 0: + entry.Kind = "symlink" + linkTarget, linkErr := os.Readlink(currentPath) + if linkErr != nil { + return fmt.Errorf("read symlink %s: %w", currentPath, linkErr) + } + entry.LinkTarget = linkTarget + case info.Mode().IsRegular(): + entry.Kind = "file" + plan.LogicalSize += info.Size() + file, openErr := os.Open(currentPath) + if openErr != nil { + return fmt.Errorf("open source file %s: %w", currentPath, openErr) + } + splitErr := s.chunker.Split(ctx, file, func(raw []byte) error { + chunkID := s.chunkID(raw, compression, task.Encrypt) + entry.Chunks = append(entry.Chunks, chunkID) + if _, exists := plan.chunks[chunkID]; exists { + return nil + } + written, writeErr := spool.Write(raw) + if writeErr != nil { + return fmt.Errorf("write repository chunk spool: %w", writeErr) + } + if written != len(raw) { + return io.ErrShortWrite + } + plan.chunks[chunkID] = repositoryPlanChunk{Offset: spoolOffset, Size: int64(len(raw))} + plan.chunkOrder = append(plan.chunkOrder, chunkID) + plan.UniqueSize += int64(len(raw)) + spoolOffset += int64(len(raw)) + return nil + }) + closeErr := file.Close() + if splitErr != nil { + return errors.Join(fmt.Errorf("chunk source file %s: %w", currentPath, splitErr), closeErr) + } + if closeErr != nil { + return fmt.Errorf("close source file %s: %w", currentPath, closeErr) + } + default: + writer.WriteLine(fmt.Sprintf("跳过不支持的特殊文件：%s", currentPath)) + delete(seenPaths, archiveName) + return nil + } + plan.snapshot.Entries = append(plan.snapshot.Entries, entry) + plan.Manifest.Entries = append(plan.Manifest.Entries, manifestEntry) + return nil + }) + if walkErr != nil { + return nil, fmt.Errorf("scan source path %s: %w", sourcePath, walkErr) + } + } + if err := spool.Sync(); err != nil { + return nil, fmt.Errorf("sync repository chunk spool: %w", err) + } + if err := spool.Close(); err != nil { + return nil, fmt.Errorf("close repository chunk spool: %w", err) + } + spoolClosed = true + writer.WriteLine(fmt.Sprintf("CDC 扫描完成：%d 个条目，逻辑数据 %d bytes，任务内唯一数据 %d bytes", len(plan.snapshot.Entries), plan.LogicalSize, plan.UniqueSize)) + return plan, nil +} + +func (s *RepositoryStore) Upload(ctx context.Context, provider storage.StorageProvider, plan *RepositoryPlan, snapshotKey string) (*RepositoryUploadResult, error) { + if provider == nil || plan == nil || strings.TrimSpace(snapshotKey) == "" { + return nil, fmt.Errorf("repository provider, plan and snapshot key are required") + } + locations, err := s.loadIndex(ctx, provider) + if err != nil { + return nil, err + } + spool, err := os.Open(plan.spoolPath) + if err != nil { + return nil, fmt.Errorf("open repository chunk spool: %w", err) + } + uploadedBytes, newChunks, err := s.uploadMissingChunks(ctx, provider, plan, spool, locations) + closeErr := spool.Close() + if err != nil { + return nil, errors.Join(err, closeErr) + } + if closeErr != nil { + return nil, fmt.Errorf("close repository chunk spool: %w", closeErr) + } + + snapshotBytes, err := s.encodeSnapshot(plan.snapshot) + if err != nil { + return nil, err + } + if err := provider.Upload(ctx, snapshotKey, bytes.NewReader(snapshotBytes), int64(len(snapshotBytes)), map[string]string{"format": "backupx-repository-v1"}); err != nil { + return nil, fmt.Errorf("upload repository snapshot %s: %w", snapshotKey, err) + } + digest := sha256.Sum256(snapshotBytes) + reusedBytes := int64(0) + for _, chunkID := range plan.chunkOrder { + if _, wasNew := newChunks[chunkID]; !wasNew { + reusedBytes += plan.chunks[chunkID].Size + } + } + return &RepositoryUploadResult{ + SnapshotKey: snapshotKey, + SnapshotSize: int64(len(snapshotBytes)), + LogicalSize: plan.LogicalSize, + UploadedBytes: uploadedBytes + int64(len(snapshotBytes)), + ReusedBytes: reusedBytes, + UniqueChunks: len(plan.chunkOrder), + NewChunks: len(newChunks), + Checksum: hex.EncodeToString(digest[:]), + }, nil +} + +// EstimateUploadSize returns a target-aware soft-quota estimate. Existing +// chunks are excluded so a nearly full repository can still accept a snapshot +// that is almost entirely deduplicated. +func (s *RepositoryStore) EstimateUploadSize(ctx context.Context, provider storage.StorageProvider, plan *RepositoryPlan) (int64, error) { + if provider == nil || plan == nil { + return 0, fmt.Errorf("repository provider and plan are required") + } + locations, err := s.loadIndex(ctx, provider) + if err != nil { + return 0, err + } + snapshotBytes, err := s.encodeSnapshot(plan.snapshot) + if err != nil { + return 0, err + } + estimate := int64(len(snapshotBytes)) + for _, chunkID := range plan.chunkOrder { + if _, exists := locations[chunkID]; exists { + continue + } + // Compression generally lowers this value. The small allowance covers + // encryption tags and index metadata while keeping the check conservative. + estimate += plan.chunks[chunkID].Size + 256 + } + return estimate, nil +} + +func (s *RepositoryStore) Restore(ctx context.Context, provider storage.StorageProvider, snapshotKey string, task TaskSpec, writer LogWriter) error { + if writer == nil { + writer = NopLogWriter{} + } + snapshot, _, err := s.loadSnapshot(ctx, provider, snapshotKey) + if err != nil { + return err + } + locations, err := s.loadIndex(ctx, provider) + if err != nil { + return err + } + restoreSource := strings.TrimSpace(task.SourcePath) + if len(task.SourcePaths) > 0 { + restoreSource = strings.TrimSpace(task.SourcePaths[0]) + } + if restoreSource == "" && len(snapshot.SourcePaths) > 0 { + restoreSource = snapshot.SourcePaths[0] + } + targetRoot := filepath.Dir(filepath.Clean(restoreSource)) + if strings.TrimSpace(task.RestoreTargetPath) != "" { + targetRoot = filepath.Clean(task.RestoreTargetPath) + } + if !filepath.IsAbs(targetRoot) { + return fmt.Errorf("repository restore target must be absolute: %s", targetRoot) + } + if err := os.MkdirAll(targetRoot, 0o755); err != nil { + return fmt.Errorf("create repository restore root: %w", err) + } + + restored := 0 + directories := make([]repositoryEntry, 0) + for _, entry := range snapshot.Entries { + if err := ctx.Err(); err != nil { + return err + } + if len(task.SelectedPaths) > 0 && !pathSelected(entry.Path, task.SelectedPaths) { + continue + } + targetPath, ok := resolveWithinParent(targetRoot, entry.Path) + if !ok { + return fmt.Errorf("unsafe repository restore path %q", entry.Path) + } + if err := s.ensureSafeParent(targetRoot, targetPath); err != nil { + return err + } + switch entry.Kind { + case "directory": + if info, statErr := os.Lstat(targetPath); statErr == nil { + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("restore directory crosses existing symlink: %s", targetPath) + } + } else if !errors.Is(statErr, os.ErrNotExist) { + return fmt.Errorf("inspect restore directory %s: %w", targetPath, statErr) + } + if err := os.MkdirAll(targetPath, os.FileMode(entry.Mode)); err != nil { + return fmt.Errorf("create restore directory %s: %w", targetPath, err) + } + directories = append(directories, entry) + case "symlink": + if err := os.RemoveAll(targetPath); err != nil { + return fmt.Errorf("replace restore symlink %s: %w", targetPath, err) + } + if err := os.Symlink(entry.LinkTarget, targetPath); err != nil { + return fmt.Errorf("create restore symlink %s: %w", targetPath, err) + } + case "file": + if info, statErr := os.Lstat(targetPath); statErr == nil { + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("restore file would overwrite existing symlink: %s", targetPath) + } + } else if !errors.Is(statErr, os.ErrNotExist) { + return fmt.Errorf("inspect restore file %s: %w", targetPath, statErr) + } + file, openErr := os.OpenFile(targetPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, os.FileMode(entry.Mode)) + if openErr != nil { + return fmt.Errorf("create restore file %s: %w", targetPath, openErr) + } + var written int64 + for _, chunkID := range entry.Chunks { + raw, readErr := s.readChunk(ctx, provider, chunkID, locations) + if readErr != nil { + return errors.Join(fmt.Errorf("restore %s: %w", entry.Path, readErr), file.Close()) + } + count, writeErr := file.Write(raw) + written += int64(count) + if writeErr != nil { + return errors.Join(fmt.Errorf("write restore file %s: %w", targetPath, writeErr), file.Close()) + } + if count != len(raw) { + return errors.Join(io.ErrShortWrite, file.Close()) + } + } + if closeErr := file.Close(); closeErr != nil { + return fmt.Errorf("close restore file %s: %w", targetPath, closeErr) + } + if written != entry.Size { + return fmt.Errorf("restored size mismatch for %s: expected %d, got %d", entry.Path, entry.Size, written) + } + if err := os.Chmod(targetPath, os.FileMode(entry.Mode)); err != nil { + return fmt.Errorf("restore mode for %s: %w", targetPath, err) + } + default: + return fmt.Errorf("unsupported repository entry kind %q", entry.Kind) + } + if entry.Kind == "file" { + modTime := time.Unix(0, entry.ModTime) + if err := os.Chtimes(targetPath, modTime, modTime); err != nil { + return fmt.Errorf("restore timestamp for %s: %w", targetPath, err) + } + } + restored++ + } + // Children modify their parent directory timestamps, so directory metadata + // is restored only after every selected file has been written. + for index := len(directories) - 1; index >= 0; index-- { + entry := directories[index] + targetPath, ok := resolveWithinParent(targetRoot, entry.Path) + if !ok { + return fmt.Errorf("unsafe repository restore path %q", entry.Path) + } + if err := os.Chmod(targetPath, os.FileMode(entry.Mode)); err != nil { + return fmt.Errorf("restore mode for %s: %w", targetPath, err) + } + modTime := time.Unix(0, entry.ModTime) + if err := os.Chtimes(targetPath, modTime, modTime); err != nil { + return fmt.Errorf("restore timestamp for %s: %w", targetPath, err) + } + } + writer.WriteLine(fmt.Sprintf("CDC 仓库恢复完成：%d 个条目", restored)) + return nil +} + +func (s *RepositoryStore) ExportTar(ctx context.Context, provider storage.StorageProvider, snapshotKey, destination string) (err error) { + snapshot, _, err := s.loadSnapshot(ctx, provider, snapshotKey) + if err != nil { + return err + } + locations, err := s.loadIndex(ctx, provider) + if err != nil { + return err + } + file, err := os.Create(destination) + if err != nil { + return fmt.Errorf("create repository export: %w", err) + } + tw := tar.NewWriter(file) + tarClosed := false + fileClosed := false + defer func() { + if !tarClosed { + err = errors.Join(err, tw.Close()) + } + if !fileClosed { + err = errors.Join(err, file.Close()) + } + }() + for _, entry := range snapshot.Entries { + if err := ctx.Err(); err != nil { + return err + } + header := &tar.Header{ + Name: entry.Path, + Mode: int64(entry.Mode), + ModTime: time.Unix(0, entry.ModTime), + Size: entry.Size, + } + switch entry.Kind { + case "directory": + header.Typeflag = tar.TypeDir + header.Size = 0 + header.Name = strings.TrimSuffix(entry.Path, "/") + "/" + case "symlink": + header.Typeflag = tar.TypeSymlink + header.Size = 0 + header.Linkname = entry.LinkTarget + case "file": + header.Typeflag = tar.TypeReg + default: + return fmt.Errorf("unsupported repository entry kind %q", entry.Kind) + } + if err := tw.WriteHeader(header); err != nil { + return fmt.Errorf("write repository tar header: %w", err) + } + if entry.Kind == "file" { + for _, chunkID := range entry.Chunks { + raw, readErr := s.readChunk(ctx, provider, chunkID, locations) + if readErr != nil { + return readErr + } + if _, writeErr := tw.Write(raw); writeErr != nil { + return fmt.Errorf("write repository tar data: %w", writeErr) + } + } + } + } + closeErr := tw.Close() + tarClosed = true + if closeErr != nil { + return fmt.Errorf("close repository tar: %w", closeErr) + } + closeErr = file.Close() + fileClosed = true + if closeErr != nil { + return fmt.Errorf("close repository export: %w", closeErr) + } + return nil +} + +func (s *RepositoryStore) Verify(ctx context.Context, provider storage.StorageProvider, snapshotKey, expectedChecksum string) (*RepositoryVerifyResult, error) { + snapshot, snapshotBytes, err := s.loadSnapshot(ctx, provider, snapshotKey) + if err != nil { + return nil, err + } + if strings.TrimSpace(expectedChecksum) != "" { + digest := sha256.Sum256(snapshotBytes) + if !strings.EqualFold(hex.EncodeToString(digest[:]), strings.TrimSpace(expectedChecksum)) { + return nil, fmt.Errorf("repository snapshot checksum mismatch") + } + } + locations, err := s.loadIndex(ctx, provider) + if err != nil { + return nil, err + } + seen := make(map[string]struct{}) + result := &RepositoryVerifyResult{Entries: len(snapshot.Entries)} + for _, entry := range snapshot.Entries { + for _, chunkID := range entry.Chunks { + if _, checked := seen[chunkID]; checked { + continue + } + raw, readErr := s.readChunk(ctx, provider, chunkID, locations) + if readErr != nil { + return nil, readErr + } + seen[chunkID] = struct{}{} + result.Chunks++ + result.Bytes += int64(len(raw)) + } + } + return result, nil +} + +func (s *RepositoryStore) Prune(ctx context.Context, provider storage.StorageProvider) (*RepositoryPruneResult, error) { + if provider == nil { + return nil, fmt.Errorf("repository provider is required") + } + liveChunks := make(map[string]struct{}) + snapshots, err := provider.List(ctx, repositorySnapshotRoot) + if err != nil { + return nil, fmt.Errorf("list repository snapshots: %w", err) + } + sort.Slice(snapshots, func(i, j int) bool { return snapshots[i].Key < snapshots[j].Key }) + for _, object := range snapshots { + snapshot, _, loadErr := s.loadSnapshot(ctx, provider, object.Key) + if loadErr != nil { + return nil, fmt.Errorf("refuse to prune with unreadable snapshot %s: %w", object.Key, loadErr) + } + for _, entry := range snapshot.Entries { + for _, chunkID := range entry.Chunks { + liveChunks[chunkID] = struct{}{} + } + } + } + + indexObjects, err := provider.List(ctx, repositoryIndexPrefix) + if err != nil { + return nil, fmt.Errorf("list repository indexes: %w", err) + } + packSizes := make(map[string]int64) + packObjects, err := provider.List(ctx, repositoryPackPrefix) + if err != nil { + return nil, fmt.Errorf("list repository packs: %w", err) + } + for _, object := range packObjects { + packSizes[object.Key] = object.Size + } + keptPacks := make(map[string]struct{}) + result := &RepositoryPruneResult{} + for _, object := range indexObjects { + segment, readErr := s.readIndexSegment(ctx, provider, object.Key) + if readErr != nil { + return nil, fmt.Errorf("refuse to prune with unreadable index %s: %w", object.Key, readErr) + } + keep := false + for chunkID := range segment.Chunks { + if _, live := liveChunks[chunkID]; live { + keep = true + break + } + } + if keep { + keptPacks[segment.Pack] = struct{}{} + continue + } + if err := provider.Delete(ctx, object.Key); err != nil { + return nil, fmt.Errorf("delete unused repository index %s: %w", object.Key, err) + } + result.DeletedIndexes++ + if err := provider.Delete(ctx, segment.Pack); err != nil { + return nil, fmt.Errorf("delete unused repository pack %s: %w", segment.Pack, err) + } + result.DeletedPacks++ + result.ReclaimedBytes += packSizes[segment.Pack] + delete(packSizes, segment.Pack) + } + for packKey, size := range packSizes { + if _, keep := keptPacks[packKey]; keep { + continue + } + if err := provider.Delete(ctx, packKey); err != nil { + return nil, fmt.Errorf("delete orphaned repository pack %s: %w", packKey, err) + } + result.DeletedPacks++ + result.ReclaimedBytes += size + } + return result, nil +} + +func (s *RepositoryStore) uploadMissingChunks(ctx context.Context, provider storage.StorageProvider, plan *RepositoryPlan, spool *os.File, locations map[string]repositoryChunkLocation) (uploadedBytes int64, newChunkIDs map[string]struct{}, err error) { + newChunkIDs = make(map[string]struct{}) + packPath := "" + var packFile *os.File + packSize := int64(0) + packChunks := make(map[string]repositoryChunkLocation) + + closeAndRemove := func() error { + var cleanupErr error + if packFile != nil { + cleanupErr = packFile.Close() + packFile = nil + } + if packPath != "" { + cleanupErr = errors.Join(cleanupErr, os.Remove(packPath)) + packPath = "" + } + return cleanupErr + } + flush := func() error { + if packFile == nil || len(packChunks) == 0 { + return nil + } + if err := packFile.Sync(); err != nil { + return fmt.Errorf("sync repository pack: %w", err) + } + if err := packFile.Close(); err != nil { + return fmt.Errorf("close repository pack: %w", err) + } + packFile = nil + packKey, indexBytes, packBytes, err := s.uploadPack(ctx, provider, packPath, packChunks) + if err != nil { + return err + } + for chunkID, location := range packChunks { + location.Pack = packKey + locations[chunkID] = location + newChunkIDs[chunkID] = struct{}{} + } + uploadedBytes += packBytes + indexBytes + if err := os.Remove(packPath); err != nil { + return fmt.Errorf("remove temporary repository pack: %w", err) + } + packPath = "" + packSize = 0 + packChunks = make(map[string]repositoryChunkLocation) + return nil + } + defer func() { + if cleanupErr := closeAndRemove(); cleanupErr != nil { + err = errors.Join(err, fmt.Errorf("clean temporary repository pack: %w", cleanupErr)) + } + }() + + for _, chunkID := range plan.chunkOrder { + if _, exists := locations[chunkID]; exists { + continue + } + planChunk := plan.chunks[chunkID] + raw := make([]byte, planChunk.Size) + readCount, readErr := spool.ReadAt(raw, planChunk.Offset) + if readErr != nil && readErr != io.EOF { + return 0, nil, fmt.Errorf("read repository chunk spool: %w", readErr) + } + if int64(readCount) != planChunk.Size { + return 0, nil, io.ErrUnexpectedEOF + } + encoded, encodeErr := s.encodeChunk(raw, plan.snapshot.Compression, plan.snapshot.Encrypted, chunkID) + if encodeErr != nil { + return 0, nil, encodeErr + } + if packFile != nil && packSize > 0 && packSize+int64(len(encoded)) > s.packSize { + if err := flush(); err != nil { + return 0, nil, err + } + } + if packFile == nil { + created, createErr := os.CreateTemp(plan.tempDir, "repository-pack-*") + if createErr != nil { + return 0, nil, fmt.Errorf("create temporary repository pack: %w", createErr) + } + packFile = created + packPath = created.Name() + } + written, writeErr := packFile.Write(encoded) + if writeErr != nil { + return 0, nil, fmt.Errorf("write repository pack: %w", writeErr) + } + if written != len(encoded) { + return 0, nil, io.ErrShortWrite + } + packChunks[chunkID] = repositoryChunkLocation{ + Offset: packSize, + Length: int64(len(encoded)), + PlainSize: planChunk.Size, + Compression: plan.snapshot.Compression, + Encrypted: plan.snapshot.Encrypted, + } + packSize += int64(len(encoded)) + } + if err := flush(); err != nil { + return 0, nil, err + } + return uploadedBytes, newChunkIDs, nil +} + +func (s *RepositoryStore) uploadPack(ctx context.Context, provider storage.StorageProvider, packPath string, chunks map[string]repositoryChunkLocation) (packKey string, indexSize int64, packSize int64, err error) { + pack, err := os.Open(packPath) + if err != nil { + return "", 0, 0, fmt.Errorf("open repository pack: %w", err) + } + packClosed := false + defer func() { + if !packClosed { + err = errors.Join(err, pack.Close()) + } + }() + hasher := sha256.New() + packBytes, err := io.Copy(hasher, pack) + if err != nil { + return "", 0, 0, fmt.Errorf("hash repository pack: %w", err) + } + packID := hex.EncodeToString(hasher.Sum(nil)) + packKey = fmt.Sprintf("%s/%s/%s.pack", repositoryPackPrefix, packID[:2], packID) + for chunkID, location := range chunks { + location.Pack = packKey + chunks[chunkID] = location + } + segment := repositoryIndexSegment{Version: repositoryFormatVersion, CreatedAt: time.Now().UTC(), Pack: packKey, Chunks: chunks} + indexBytes, err := json.Marshal(segment) + if err != nil { + return "", 0, 0, fmt.Errorf("encode repository index: %w", err) + } + if _, err := pack.Seek(0, io.SeekStart); err != nil { + return "", 0, 0, fmt.Errorf("rewind repository pack: %w", err) + } + if err := provider.Upload(ctx, packKey, pack, packBytes, map[string]string{"format": "backupx-pack-v1"}); err != nil { + return "", 0, 0, fmt.Errorf("upload repository pack %s: %w", packKey, err) + } + closeErr := pack.Close() + packClosed = true + if closeErr != nil { + return "", 0, 0, fmt.Errorf("close repository pack: %w", closeErr) + } + indexKey := fmt.Sprintf("%s/%s.json", repositoryIndexPrefix, packID) + if err := provider.Upload(ctx, indexKey, bytes.NewReader(indexBytes), int64(len(indexBytes)), map[string]string{"format": "backupx-index-v1"}); err != nil { + deleteErr := provider.Delete(ctx, packKey) + return "", 0, 0, errors.Join(fmt.Errorf("upload repository index %s: %w", indexKey, err), deleteErr) + } + return packKey, int64(len(indexBytes)), packBytes, nil +} + +func (s *RepositoryStore) loadIndex(ctx context.Context, provider storage.StorageProvider) (map[string]repositoryChunkLocation, error) { + objects, err := provider.List(ctx, repositoryIndexPrefix) + if err != nil { + return nil, fmt.Errorf("list repository indexes: %w", err) + } + sort.Slice(objects, func(i, j int) bool { return objects[i].Key < objects[j].Key }) + locations := make(map[string]repositoryChunkLocation) + for _, object := range objects { + segment, readErr := s.readIndexSegment(ctx, provider, object.Key) + if readErr != nil { + return nil, readErr + } + for chunkID, location := range segment.Chunks { + if location.Pack == "" { + location.Pack = segment.Pack + } + if _, exists := locations[chunkID]; !exists { + locations[chunkID] = location + } + } + } + return locations, nil +} + +func (s *RepositoryStore) readIndexSegment(ctx context.Context, provider storage.StorageProvider, key string) (*repositoryIndexSegment, error) { + reader, err := provider.Download(ctx, key) + if err != nil { + return nil, fmt.Errorf("download repository index %s: %w", key, err) + } + data, readErr := io.ReadAll(io.LimitReader(reader, 64<<20)) + closeErr := reader.Close() + if readErr != nil { + return nil, fmt.Errorf("read repository index %s: %w", key, readErr) + } + if closeErr != nil { + return nil, fmt.Errorf("close repository index %s: %w", key, closeErr) + } + var segment repositoryIndexSegment + if err := json.Unmarshal(data, &segment); err != nil { + return nil, fmt.Errorf("decode repository index %s: %w", key, err) + } + if segment.Version != repositoryFormatVersion || strings.TrimSpace(segment.Pack) == "" { + return nil, fmt.Errorf("unsupported repository index %s", key) + } + for chunkID, location := range segment.Chunks { + if chunkID == "" || location.Offset < 0 || location.Length <= 0 || location.PlainSize < 0 { + return nil, fmt.Errorf("invalid chunk location in repository index %s", key) + } + } + return &segment, nil +} + +func (s *RepositoryStore) loadSnapshot(ctx context.Context, provider storage.StorageProvider, key string) (*repositorySnapshot, []byte, error) { + reader, err := provider.Download(ctx, key) + if err != nil { + return nil, nil, fmt.Errorf("download repository snapshot %s: %w", key, err) + } + data, readErr := io.ReadAll(io.LimitReader(reader, 256<<20)) + closeErr := reader.Close() + if readErr != nil { + return nil, nil, fmt.Errorf("read repository snapshot %s: %w", key, readErr) + } + if closeErr != nil { + return nil, nil, fmt.Errorf("close repository snapshot %s: %w", key, closeErr) + } + var envelope repositorySnapshotEnvelope + if err := json.Unmarshal(data, &envelope); err != nil { + return nil, nil, fmt.Errorf("decode repository snapshot envelope %s: %w", key, err) + } + if envelope.Version != repositoryFormatVersion { + return nil, nil, fmt.Errorf("unsupported repository snapshot version %d", envelope.Version) + } + payload := []byte(envelope.Data) + if envelope.Encrypted { + ciphertext, decodeErr := base64.RawURLEncoding.DecodeString(envelope.Ciphertext) + if decodeErr != nil { + return nil, nil, fmt.Errorf("decode repository snapshot ciphertext: %w", decodeErr) + } + payload, err = s.decrypt(ciphertext, []byte("backupx-repository-snapshot-v1")) + if err != nil { + return nil, nil, fmt.Errorf("decrypt repository snapshot: %w", err) + } + } + var snapshot repositorySnapshot + if err := json.Unmarshal(payload, &snapshot); err != nil { + return nil, nil, fmt.Errorf("decode repository snapshot: %w", err) + } + if snapshot.Version != repositoryFormatVersion { + return nil, nil, fmt.Errorf("unsupported repository snapshot payload version %d", snapshot.Version) + } + return &snapshot, data, nil +} + +func (s *RepositoryStore) encodeSnapshot(snapshot repositorySnapshot) ([]byte, error) { + payload, err := json.Marshal(snapshot) + if err != nil { + return nil, fmt.Errorf("encode repository snapshot: %w", err) + } + envelope := repositorySnapshotEnvelope{Version: repositoryFormatVersion, Encrypted: snapshot.Encrypted} + if snapshot.Encrypted { + ciphertext, encryptErr := s.encrypt(payload, []byte("backupx-repository-snapshot-v1")) + if encryptErr != nil { + return nil, fmt.Errorf("encrypt repository snapshot: %w", encryptErr) + } + envelope.Ciphertext = base64.RawURLEncoding.EncodeToString(ciphertext) + } else { + envelope.Data = payload + } + data, err := json.Marshal(envelope) + if err != nil { + return nil, fmt.Errorf("encode repository snapshot envelope: %w", err) + } + return data, nil +} + +func (s *RepositoryStore) readChunk(ctx context.Context, provider storage.StorageProvider, chunkID string, locations map[string]repositoryChunkLocation) ([]byte, error) { + location, exists := locations[chunkID] + if !exists { + return nil, fmt.Errorf("repository chunk %s is missing from the index", chunkID) + } + var reader io.ReadCloser + var err error + if ranged, ok := provider.(storage.StorageRangeDownloader); ok { + reader, err = ranged.DownloadRange(ctx, location.Pack, location.Offset, location.Length) + } else { + reader, err = provider.Download(ctx, location.Pack) + if err == nil && location.Offset > 0 { + if _, copyErr := io.CopyN(io.Discard, reader, location.Offset); copyErr != nil { + return nil, errors.Join(fmt.Errorf("seek repository pack %s: %w", location.Pack, copyErr), reader.Close()) + } + } + } + if err != nil { + return nil, fmt.Errorf("read repository pack %s: %w", location.Pack, err) + } + encoded := make([]byte, location.Length) + _, readErr := io.ReadFull(reader, encoded) + closeErr := reader.Close() + if readErr != nil { + return nil, fmt.Errorf("read repository chunk %s: %w", chunkID, readErr) + } + if closeErr != nil { + return nil, fmt.Errorf("close repository pack %s: %w", location.Pack, closeErr) + } + raw, err := s.decodeChunk(encoded, location, chunkID) + if err != nil { + return nil, err + } + if int64(len(raw)) != location.PlainSize { + return nil, fmt.Errorf("repository chunk %s size mismatch", chunkID) + } + if actual := s.chunkID(raw, location.Compression, location.Encrypted); actual != chunkID { + return nil, fmt.Errorf("repository chunk %s failed content verification", chunkID) + } + return raw, nil +} + +func (s *RepositoryStore) encodeChunk(raw []byte, compression string, encrypted bool, chunkID string) ([]byte, error) { + var encoded []byte + switch compression { + case "none": + encoded = append([]byte(nil), raw...) + case "gzip": + var buffer bytes.Buffer + writer := gzip.NewWriter(&buffer) + if _, err := writer.Write(raw); err != nil { + return nil, errors.Join(fmt.Errorf("gzip repository chunk: %w", err), writer.Close()) + } + if err := writer.Close(); err != nil { + return nil, fmt.Errorf("close repository gzip chunk: %w", err) + } + encoded = buffer.Bytes() + case "zstd": + writer, err := zstd.NewWriter(nil) + if err != nil { + return nil, fmt.Errorf("create repository zstd encoder: %w", err) + } + encoded = writer.EncodeAll(raw, nil) + writer.Close() + default: + return nil, fmt.Errorf("unsupported repository compression %q", compression) + } + if !encrypted { + return encoded, nil + } + ciphertext, err := s.encrypt(encoded, []byte(chunkID)) + if err != nil { + return nil, fmt.Errorf("encrypt repository chunk: %w", err) + } + return ciphertext, nil +} + +func (s *RepositoryStore) decodeChunk(encoded []byte, location repositoryChunkLocation, chunkID string) ([]byte, error) { + payload := encoded + var err error + if location.Encrypted { + payload, err = s.decrypt(encoded, []byte(chunkID)) + if err != nil { + return nil, fmt.Errorf("decrypt repository chunk %s: %w", chunkID, err) + } + } + switch location.Compression { + case "none": + return payload, nil + case "gzip": + reader, err := gzip.NewReader(bytes.NewReader(payload)) + if err != nil { + return nil, fmt.Errorf("open repository gzip chunk: %w", err) + } + raw, readErr := io.ReadAll(reader) + closeErr := reader.Close() + if readErr != nil { + return nil, fmt.Errorf("decompress repository gzip chunk: %w", readErr) + } + if closeErr != nil { + return nil, fmt.Errorf("close repository gzip chunk: %w", closeErr) + } + return raw, nil + case "zstd": + reader, err := zstd.NewReader(nil) + if err != nil { + return nil, fmt.Errorf("create repository zstd decoder: %w", err) + } + raw, err := reader.DecodeAll(payload, nil) + reader.Close() + if err != nil { + return nil, fmt.Errorf("decompress repository zstd chunk: %w", err) + } + return raw, nil + default: + return nil, fmt.Errorf("unsupported repository compression %q", location.Compression) + } +} + +func (s *RepositoryStore) chunkID(raw []byte, compression string, encrypted bool) string { + domain := fmt.Sprintf("backupx-repository-v1/%s/plain\x00", compression) + prefix := "p-" + if encrypted { + domain = fmt.Sprintf("backupx-repository-v1/%s/encrypted\x00", compression) + mac := hmac.New(sha256.New, s.key) + mac.Write([]byte(domain)) + mac.Write(raw) + return "e-" + hex.EncodeToString(mac.Sum(nil)) + } + digest := sha256.New() + digest.Write([]byte(domain)) + digest.Write(raw) + return prefix + hex.EncodeToString(digest.Sum(nil)) +} + +func (s *RepositoryStore) encrypt(plain, additionalData []byte) ([]byte, error) { + if len(s.key) != 32 { + return nil, fmt.Errorf("repository encryption key is unavailable") + } + block, err := aes.NewCipher(s.key) + if err != nil { + return nil, err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return nil, err + } + return gcm.Seal(nonce, nonce, plain, additionalData), nil +} + +func (s *RepositoryStore) decrypt(ciphertext, additionalData []byte) ([]byte, error) { + if len(s.key) != 32 { + return nil, fmt.Errorf("repository encryption key is unavailable") + } + block, err := aes.NewCipher(s.key) + if err != nil { + return nil, err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + if len(ciphertext) < gcm.NonceSize() { + return nil, fmt.Errorf("repository ciphertext is too short") + } + nonce := ciphertext[:gcm.NonceSize()] + return gcm.Open(nil, nonce, ciphertext[gcm.NonceSize():], additionalData) +} + +func (s *RepositoryStore) normalizeCompression(value string) (string, error) { + switch strings.ToLower(strings.TrimSpace(value)) { + case "", "zstd": + return "zstd", nil + case "gzip": + return "gzip", nil + case "none": + return "none", nil + default: + return "", fmt.Errorf("unsupported repository compression %q", value) + } +} + +func (s *RepositoryStore) ensureSafeParent(root, target string) error { + parent := filepath.Dir(target) + relative, err := filepath.Rel(root, parent) + if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { + return fmt.Errorf("restore target escapes repository root: %s", target) + } + current := root + if relative == "." { + return nil + } + for _, segment := range strings.Split(relative, string(filepath.Separator)) { + current = filepath.Join(current, segment) + info, statErr := os.Lstat(current) + if errors.Is(statErr, os.ErrNotExist) { + if err := os.Mkdir(current, 0o755); err != nil && !errors.Is(err, os.ErrExist) { + return fmt.Errorf("create restore parent %s: %w", current, err) + } + continue + } + if statErr != nil { + return fmt.Errorf("inspect restore parent %s: %w", current, statErr) + } + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return fmt.Errorf("unsafe restore parent %s", current) + } + } + return nil +} + +func compactPaths(items []string) []string { + result := make([]string, 0, len(items)) + for _, item := range items { + if value := strings.TrimSpace(item); value != "" { + result = append(result, filepath.Clean(value)) + } + } + return result +} diff --git a/server/internal/backup/repository_chunker.go b/server/internal/backup/repository_chunker.go new file mode 100644 index 0000000..1dabf21 --- /dev/null +++ b/server/internal/backup/repository_chunker.go @@ -0,0 +1,131 @@ +package backup + +import ( + "context" + "fmt" + "io" +) + +const ( + repositoryChunkMin = 512 << 10 + repositoryChunkAvg = 1 << 20 + repositoryChunkMax = 4 << 20 +) + +// contentDefinedChunker implements the normalized FastCDC cut-point strategy. +// The rolling Gear hash only retains the latest 64 bytes through uint64 +// overflow, so chunk boundaries re-synchronize after insertions or deletions. +type contentDefinedChunker struct { + minSize int + avgSize int + maxSize int + smallMask uint64 + largeMask uint64 + gear [256]uint64 +} + +func newContentDefinedChunker() *contentDefinedChunker { + chunker := &contentDefinedChunker{ + minSize: repositoryChunkMin, + avgSize: repositoryChunkAvg, + maxSize: repositoryChunkMax, + smallMask: (1 << 21) - 1, + largeMask: (1 << 19) - 1, + } + + // SplitMix64 produces a stable, well-distributed Gear table. The seed and + // generation algorithm are part of repository format v1 and must not change. + seed := uint64(0x6a09e667f3bcc909) + for i := range chunker.gear { + seed += 0x9e3779b97f4a7c15 + value := seed + value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9 + value = (value ^ (value >> 27)) * 0x94d049bb133111eb + chunker.gear[i] = value ^ (value >> 31) + } + return chunker +} + +func (c *contentDefinedChunker) Split(ctx context.Context, reader io.Reader, emit func([]byte) error) error { + if reader == nil || emit == nil { + return fmt.Errorf("chunk reader and emitter are required") + } + + pending := make([]byte, 0, c.maxSize+(256<<10)) + readBuffer := make([]byte, 256<<10) + eof := false + for { + if !eof { + readCount, readErr := reader.Read(readBuffer) + if readCount > 0 { + pending = append(pending, readBuffer[:readCount]...) + } + switch readErr { + case nil: + case io.EOF: + eof = true + default: + return fmt.Errorf("read source for chunking: %w", readErr) + } + if readCount == 0 && readErr == nil { + continue + } + } + + for len(pending) > 0 { + if err := ctx.Err(); err != nil { + return err + } + cut := c.findCutPoint(pending, eof) + if cut == 0 { + break + } + chunk := make([]byte, cut) + copy(chunk, pending[:cut]) + if err := emit(chunk); err != nil { + return err + } + copy(pending, pending[cut:]) + pending = pending[:len(pending)-cut] + } + + if eof { + if len(pending) != 0 { + return fmt.Errorf("chunker stopped with %d buffered bytes", len(pending)) + } + return nil + } + } +} + +func (c *contentDefinedChunker) findCutPoint(data []byte, eof bool) int { + if len(data) < c.minSize { + if eof { + return len(data) + } + return 0 + } + + limit := len(data) + if limit > c.maxSize { + limit = c.maxSize + } + var hash uint64 + for index := c.minSize; index < limit; index++ { + hash = (hash << 1) + c.gear[data[index]] + mask := c.largeMask + if index < c.avgSize { + mask = c.smallMask + } + if hash&mask == 0 { + return index + 1 + } + } + if len(data) >= c.maxSize { + return c.maxSize + } + if eof { + return len(data) + } + return 0 +} diff --git a/server/internal/backup/repository_test.go b/server/internal/backup/repository_test.go new file mode 100644 index 0000000..c17d670 --- /dev/null +++ b/server/internal/backup/repository_test.go @@ -0,0 +1,260 @@ +package backup + +import ( + "bytes" + "context" + "crypto/sha256" + "fmt" + "io" + "math/rand" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "backupx/server/internal/storage" +) + +func TestContentDefinedChunkerResynchronizesAfterInsertion(t *testing.T) { + source := make([]byte, 8<<20) + if _, err := rand.New(rand.NewSource(42)).Read(source); err != nil { + t.Fatalf("generate source: %v", err) + } + modified := make([]byte, 0, len(source)+4096) + modified = append(modified, source[:2<<20]...) + modified = append(modified, bytes.Repeat([]byte("inserted"), 512)...) + modified = append(modified, source[2<<20:]...) + + chunker := newContentDefinedChunker() + collect := func(data []byte) map[string]struct{} { + t.Helper() + ids := make(map[string]struct{}) + err := chunker.Split(context.Background(), bytes.NewReader(data), func(chunk []byte) error { + digest := sha256.Sum256(chunk) + ids[fmt.Sprintf("%x", digest[:])] = struct{}{} + if len(chunk) > repositoryChunkMax { + return fmt.Errorf("chunk exceeds maximum: %d", len(chunk)) + } + return nil + }) + if err != nil { + t.Fatalf("split chunks: %v", err) + } + return ids + } + originalChunks := collect(source) + modifiedChunks := collect(modified) + shared := 0 + for chunkID := range originalChunks { + if _, ok := modifiedChunks[chunkID]; ok { + shared++ + } + } + if shared < len(originalChunks)/2 { + t.Fatalf("content-defined boundaries did not resynchronize: shared=%d original=%d", shared, len(originalChunks)) + } +} + +func TestRepositoryRoundTripDedupAndPrune(t *testing.T) { + ctx := context.Background() + tempDir := t.TempDir() + sourceDir := filepath.Join(tempDir, "dataset") + if err := os.MkdirAll(filepath.Join(sourceDir, "empty"), 0o755); err != nil { + t.Fatalf("create source: %v", err) + } + original := make([]byte, 6<<20) + if _, err := rand.New(rand.NewSource(7)).Read(original); err != nil { + t.Fatalf("generate fixture: %v", err) + } + primaryPath := filepath.Join(sourceDir, "primary.bin") + duplicatePath := filepath.Join(sourceDir, "duplicate.bin") + if err := os.WriteFile(primaryPath, original, 0o640); err != nil { + t.Fatalf("write primary: %v", err) + } + if err := os.WriteFile(duplicatePath, original, 0o640); err != nil { + t.Fatalf("write duplicate: %v", err) + } + + key := sha256.Sum256([]byte("repository-test-key")) + store := NewRepositoryStore(key[:]) + provider := newMemoryRepositoryProvider() + task := TaskSpec{ + ID: 12, + Name: "repository-test", + Type: "file", + SourcePaths: []string{sourceDir}, + Compression: "zstd", + Encrypt: true, + StartedAt: time.Date(2026, 8, 6, 1, 2, 3, 0, time.UTC), + TempDir: tempDir, + } + + firstPlan, err := store.BuildPlan(ctx, task, NopLogWriter{}) + if err != nil { + t.Fatalf("build first plan: %v", err) + } + firstKey := store.SnapshotKey(task.ID, 1, task.StartedAt) + firstResult, err := store.Upload(ctx, provider, firstPlan, firstKey) + if closeErr := firstPlan.Close(); closeErr != nil { + t.Fatalf("close first plan: %v", closeErr) + } + if err != nil { + t.Fatalf("upload first snapshot: %v", err) + } + if firstResult.NewChunks == 0 || firstResult.UniqueChunks == 0 { + t.Fatalf("first upload did not create chunks: %+v", firstResult) + } + if firstPlan.UniqueSize >= firstPlan.LogicalSize { + t.Fatalf("duplicate file was not deduplicated within plan: unique=%d logical=%d", firstPlan.UniqueSize, firstPlan.LogicalSize) + } + + modified := make([]byte, 0, len(original)+4096) + modified = append(modified, original[:2<<20]...) + modified = append(modified, bytes.Repeat([]byte("changed!"), 512)...) + modified = append(modified, original[2<<20:]...) + if err := os.WriteFile(primaryPath, modified, 0o640); err != nil { + t.Fatalf("modify primary: %v", err) + } + task.StartedAt = task.StartedAt.Add(time.Hour) + secondPlan, err := store.BuildPlan(ctx, task, NopLogWriter{}) + if err != nil { + t.Fatalf("build second plan: %v", err) + } + secondKey := store.SnapshotKey(task.ID, 2, task.StartedAt) + secondResult, err := store.Upload(ctx, provider, secondPlan, secondKey) + if closeErr := secondPlan.Close(); closeErr != nil { + t.Fatalf("close second plan: %v", closeErr) + } + if err != nil { + t.Fatalf("upload second snapshot: %v", err) + } + if secondResult.ReusedBytes <= secondResult.LogicalSize/3 { + t.Fatalf("second snapshot reused too little data: %+v", secondResult) + } + if secondResult.UploadedBytes >= secondResult.LogicalSize { + t.Fatalf("incremental upload was not smaller than logical data: %+v", secondResult) + } + + verify, err := store.Verify(ctx, provider, secondKey, secondResult.Checksum) + if err != nil { + t.Fatalf("verify repository: %v", err) + } + if verify.Chunks == 0 || verify.Bytes == 0 { + t.Fatalf("empty verification result: %+v", verify) + } + + restoreRoot := filepath.Join(tempDir, "restore") + restoreTask := task + restoreTask.RestoreTargetPath = restoreRoot + if err := store.Restore(ctx, provider, secondKey, restoreTask, NopLogWriter{}); err != nil { + t.Fatalf("restore snapshot: %v", err) + } + restored, err := os.ReadFile(filepath.Join(restoreRoot, filepath.Base(sourceDir), "primary.bin")) + if err != nil { + t.Fatalf("read restored primary: %v", err) + } + if !bytes.Equal(restored, modified) { + t.Fatalf("restored primary differs from source") + } + if info, err := os.Stat(filepath.Join(restoreRoot, filepath.Base(sourceDir), "empty")); err != nil || !info.IsDir() { + t.Fatalf("empty directory was not restored: info=%v err=%v", info, err) + } + + if err := provider.Delete(ctx, firstKey); err != nil { + t.Fatalf("delete first snapshot: %v", err) + } + if _, err := store.Prune(ctx, provider); err != nil { + t.Fatalf("prune with live snapshot: %v", err) + } + if err := provider.Delete(ctx, secondKey); err != nil { + t.Fatalf("delete second snapshot: %v", err) + } + pruned, err := store.Prune(ctx, provider) + if err != nil { + t.Fatalf("prune empty repository: %v", err) + } + if pruned.DeletedPacks == 0 || pruned.DeletedIndexes == 0 { + t.Fatalf("prune did not reclaim repository data: %+v", pruned) + } + if objects, err := provider.List(ctx, repositoryPackPrefix); err != nil || len(objects) != 0 { + t.Fatalf("packs remain after prune: objects=%v err=%v", objects, err) + } +} + +type memoryRepositoryProvider struct { + mu sync.RWMutex + objects map[string][]byte + times map[string]time.Time +} + +func newMemoryRepositoryProvider() *memoryRepositoryProvider { + return &memoryRepositoryProvider{objects: make(map[string][]byte), times: make(map[string]time.Time)} +} + +func (p *memoryRepositoryProvider) Type() storage.ProviderType { return "memory" } +func (p *memoryRepositoryProvider) TestConnection(context.Context) error { return nil } + +func (p *memoryRepositoryProvider) Upload(_ context.Context, key string, reader io.Reader, size int64, _ map[string]string) error { + data, err := io.ReadAll(reader) + if err != nil { + return err + } + if int64(len(data)) != size { + return fmt.Errorf("size mismatch for %s: %d != %d", key, len(data), size) + } + p.mu.Lock() + defer p.mu.Unlock() + p.objects[key] = append([]byte(nil), data...) + p.times[key] = time.Now().UTC() + return nil +} + +func (p *memoryRepositoryProvider) Download(_ context.Context, key string) (io.ReadCloser, error) { + p.mu.RLock() + defer p.mu.RUnlock() + data, ok := p.objects[key] + if !ok { + return nil, fmt.Errorf("object %s not found", key) + } + return io.NopCloser(bytes.NewReader(append([]byte(nil), data...))), nil +} + +func (p *memoryRepositoryProvider) DownloadRange(_ context.Context, key string, offset, length int64) (io.ReadCloser, error) { + p.mu.RLock() + defer p.mu.RUnlock() + data, ok := p.objects[key] + if !ok { + return nil, fmt.Errorf("object %s not found", key) + } + if offset < 0 || length <= 0 || offset+length > int64(len(data)) { + return nil, fmt.Errorf("invalid range %d:%d for %s", offset, length, key) + } + return io.NopCloser(bytes.NewReader(append([]byte(nil), data[offset:offset+length]...))), nil +} + +func (p *memoryRepositoryProvider) Delete(_ context.Context, key string) error { + p.mu.Lock() + defer p.mu.Unlock() + if _, ok := p.objects[key]; !ok { + return fmt.Errorf("object %s not found", key) + } + delete(p.objects, key) + delete(p.times, key) + return nil +} + +func (p *memoryRepositoryProvider) List(_ context.Context, prefix string) ([]storage.ObjectInfo, error) { + p.mu.RLock() + defer p.mu.RUnlock() + result := make([]storage.ObjectInfo, 0) + for key, data := range p.objects { + if strings.HasPrefix(key, prefix) { + result = append(result, storage.ObjectInfo{Key: key, Size: int64(len(data)), UpdatedAt: p.times[key]}) + } + } + sort.Slice(result, func(i, j int) bool { return result[i].Key < result[j].Key }) + return result, nil +} diff --git a/server/internal/backup/retention/service.go b/server/internal/backup/retention/service.go index 70dda96..6e632c9 100644 --- a/server/internal/backup/retention/service.go +++ b/server/internal/backup/retention/service.go @@ -2,11 +2,13 @@ package retention import ( "context" + "encoding/json" "fmt" "sort" "strings" "time" + "backupx/server/internal/backup" "backupx/server/internal/model" "backupx/server/internal/repository" "backupx/server/internal/storage" @@ -40,16 +42,48 @@ type CleanupResult struct { Warnings []string } -type Service struct { - records repository.BackupRecordRepository - now func() time.Time +type cleanupObject struct { + targetID uint + path string } -func NewService(records repository.BackupRecordRepository) *Service { - return &Service{records: records, now: func() time.Time { return time.Now().UTC() }} +type storedUploadResult struct { + StorageTargetID uint `json:"storageTargetId"` + Status string `json:"status"` + StoragePath string `json:"storagePath"` +} + +type Service struct { + records repository.BackupRecordRepository + repositoryKey []byte + now func() time.Time +} + +func NewService(records repository.BackupRecordRepository, repositoryKey ...[]byte) *Service { + var key []byte + if len(repositoryKey) > 0 { + key = append([]byte(nil), repositoryKey[0]...) + } + return &Service{records: records, repositoryKey: key, now: func() time.Time { return time.Now().UTC() }} } func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider storage.StorageProvider) (*CleanupResult, error) { + return s.cleanup(ctx, task, func(uint) (storage.StorageProvider, bool) { + return provider, provider != nil + }) +} + +// CleanupProviders applies one retention decision to every successful copy of +// a record before deleting its database row. This prevents multi-target tasks +// from leaving stale objects after the first target removes the shared record. +func (s *Service) CleanupProviders(ctx context.Context, task *model.BackupTask, providers map[uint]storage.StorageProvider) (*CleanupResult, error) { + return s.cleanup(ctx, task, func(targetID uint) (storage.StorageProvider, bool) { + provider, ok := providers[targetID] + return provider, ok && provider != nil + }) +} + +func (s *Service) cleanup(ctx context.Context, task *model.BackupTask, resolveProvider func(uint) (storage.StorageProvider, bool)) (*CleanupResult, error) { if task == nil { return nil, fmt.Errorf("backup task is required") } @@ -67,17 +101,35 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider // 差异链保护：保留仍被存活差异依赖的全量，避免删除基线后差异无法恢复。 candidates = protectDifferentialBases(records, candidates) result := &CleanupResult{} + repositoryProviders := make(map[uint]storage.StorageProvider) + touchedProviders := make(map[uint]storage.StorageProvider) for _, record := range candidates { - if strings.TrimSpace(record.StoragePath) != "" { - if provider == nil { - result.Warnings = append(result.Warnings, fmt.Sprintf("record %d missing storage provider for cleanup", record.ID)) + objects, objectErr := cleanupObjectsForRecord(record) + if objectErr != nil { + result.Warnings = append(result.Warnings, fmt.Sprintf("decode storage copies for record %d failed: %v", record.ID, objectErr)) + continue + } + allObjectsDeleted := true + for _, object := range objects { + provider, ok := resolveProvider(object.targetID) + if !ok { + result.Warnings = append(result.Warnings, fmt.Sprintf("record %d missing storage provider %d for cleanup", record.ID, object.targetID)) + allObjectsDeleted = false continue } - if err := provider.Delete(ctx, record.StoragePath); err != nil { - result.Warnings = append(result.Warnings, fmt.Sprintf("delete storage object %s failed: %v", record.StoragePath, err)) + if err := provider.Delete(ctx, object.path); err != nil { + result.Warnings = append(result.Warnings, fmt.Sprintf("delete storage object %s from target %d failed: %v", object.path, object.targetID, err)) + allObjectsDeleted = false continue } result.DeletedObjects++ + touchedProviders[object.targetID] = provider + if record.BackupKind == model.BackupKindRepository { + repositoryProviders[object.targetID] = provider + } + } + if !allObjectsDeleted { + continue } if err := s.records.Delete(ctx, record.ID); err != nil { result.Warnings = append(result.Warnings, fmt.Sprintf("delete backup record %d failed: %v", record.ID, err)) @@ -85,13 +137,25 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider } result.DeletedRecords++ } + for targetID, provider := range repositoryProviders { + pruned, pruneErr := backup.NewRepositoryStore(s.repositoryKey).Prune(ctx, provider) + if pruneErr != nil { + result.Warnings = append(result.Warnings, fmt.Sprintf("prune CDC repository on target %d failed: %v", targetID, pruneErr)) + } else { + result.DeletedObjects += pruned.DeletedPacks + pruned.DeletedIndexes + } + } // 清理空目录：收集被删除文件的父目录，尝试移除空目录 - if dirCleaner, ok := provider.(storage.StorageDirCleaner); ok && result.DeletedObjects > 0 { + for targetID, provider := range touchedProviders { + dirCleaner, ok := provider.(storage.StorageDirCleaner) + if !ok { + continue + } prefixes := collectDirPrefixes(candidates) for _, prefix := range prefixes { if err := dirCleaner.RemoveEmptyDirs(ctx, prefix); err != nil { - result.Warnings = append(result.Warnings, fmt.Sprintf("cleanup empty dirs for %s: %v", prefix, err)) + result.Warnings = append(result.Warnings, fmt.Sprintf("cleanup empty dirs for %s on target %d: %v", prefix, targetID, err)) } } } @@ -99,6 +163,43 @@ func (s *Service) Cleanup(ctx context.Context, task *model.BackupTask, provider return result, nil } +func cleanupObjectsForRecord(record model.BackupRecord) ([]cleanupObject, error) { + defaultPath := strings.TrimSpace(record.StoragePath) + if strings.TrimSpace(record.StorageUploadResults) == "" { + if defaultPath == "" { + return nil, nil + } + return []cleanupObject{{targetID: record.StorageTargetID, path: defaultPath}}, nil + } + var results []storedUploadResult + if err := json.Unmarshal([]byte(record.StorageUploadResults), &results); err != nil { + return nil, err + } + objects := make([]cleanupObject, 0, len(results)) + seen := make(map[uint]struct{}, len(results)) + for _, result := range results { + if !strings.EqualFold(strings.TrimSpace(result.Status), model.BackupRecordStatusSuccess) { + continue + } + objectPath := strings.TrimSpace(result.StoragePath) + if objectPath == "" { + objectPath = defaultPath + } + if objectPath == "" { + continue + } + if _, exists := seen[result.StorageTargetID]; exists { + continue + } + seen[result.StorageTargetID] = struct{}{} + objects = append(objects, cleanupObject{targetID: result.StorageTargetID, path: objectPath}) + } + if len(objects) == 0 && defaultPath != "" { + return nil, fmt.Errorf("successful record has no successful storage copy") + } + return objects, nil +} + // protectDifferentialBases 从删除候选中剔除「仍被存活差异依赖的全量」， // 避免删除基线后其差异备份失去依据、无法恢复。全量仅当其全部差异都已过期/删除时才会被清理。 func protectDifferentialBases(all []model.BackupRecord, candidates []model.BackupRecord) []model.BackupRecord { diff --git a/server/internal/backup/retention/service_test.go b/server/internal/backup/retention/service_test.go index a1e8124..8cb8108 100644 --- a/server/internal/backup/retention/service_test.go +++ b/server/internal/backup/retention/service_test.go @@ -221,3 +221,66 @@ func TestCleanupDeletesExpiredRecords(t *testing.T) { t.Fatalf("unexpected deleted objects: %#v", provider.deleted) } } + +func TestCleanupProvidersDeletesEverySuccessfulCopyBeforeRecord(t *testing.T) { + now := time.Date(2026, 3, 7, 16, 0, 0, 0, time.UTC) + completedNew := now.Add(-time.Hour) + completedOld := now.Add(-24 * time.Hour) + repo := &fakeRecordRepository{records: []model.BackupRecord{ + {ID: 2, TaskID: 1, StoragePath: "records/2", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedNew}, + { + ID: 1, TaskID: 1, StoragePath: "records/1", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedOld, + StorageUploadResults: `[{"storageTargetId":11,"status":"success","storagePath":"first/1"},{"storageTargetId":12,"status":"success","storagePath":"second/1"},{"storageTargetId":13,"status":"failed"}]`, + }, + }} + first := &fakeProvider{} + second := &fakeProvider{} + service := NewService(repo) + service.now = func() time.Time { return now } + + result, err := service.CleanupProviders(context.Background(), &model.BackupTask{ID: 1, MaxBackups: 1}, map[uint]storage.StorageProvider{ + 11: first, + 12: second, + }) + if err != nil { + t.Fatalf("CleanupProviders returned error: %v", err) + } + if result.DeletedRecords != 1 || result.DeletedObjects != 2 || len(result.Warnings) != 0 { + t.Fatalf("unexpected cleanup result: %#v", result) + } + if len(repo.deleted) != 1 || repo.deleted[0] != 1 { + t.Fatalf("unexpected deleted records: %#v", repo.deleted) + } + if len(first.deleted) != 1 || first.deleted[0] != "first/1" { + t.Fatalf("unexpected first-target deletes: %#v", first.deleted) + } + if len(second.deleted) != 1 || second.deleted[0] != "second/1" { + t.Fatalf("unexpected second-target deletes: %#v", second.deleted) + } +} + +func TestCleanupProvidersKeepsRecordWhenCopyProviderIsUnavailable(t *testing.T) { + now := time.Date(2026, 3, 7, 16, 0, 0, 0, time.UTC) + completedNew := now.Add(-time.Hour) + completedOld := now.Add(-24 * time.Hour) + repo := &fakeRecordRepository{records: []model.BackupRecord{ + {ID: 2, TaskID: 1, StoragePath: "records/2", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedNew}, + { + ID: 1, TaskID: 1, StoragePath: "records/1", Status: model.BackupRecordStatusSuccess, CompletedAt: &completedOld, + StorageUploadResults: `[{"storageTargetId":11,"status":"success","storagePath":"first/1"},{"storageTargetId":12,"status":"success","storagePath":"second/1"}]`, + }, + }} + first := &fakeProvider{} + service := NewService(repo) + + result, err := service.CleanupProviders(context.Background(), &model.BackupTask{ID: 1, MaxBackups: 1}, map[uint]storage.StorageProvider{11: first}) + if err != nil { + t.Fatalf("CleanupProviders returned error: %v", err) + } + if result.DeletedRecords != 0 || result.DeletedObjects != 1 || len(result.Warnings) != 1 { + t.Fatalf("unexpected safe partial-cleanup result: %#v", result) + } + if len(repo.deleted) != 0 { + t.Fatalf("record must remain until all copies are deleted: %#v", repo.deleted) + } +} diff --git a/server/internal/model/backup_record.go b/server/internal/model/backup_record.go index 9f8f9c6..5f0a965 100644 --- a/server/internal/model/backup_record.go +++ b/server/internal/model/backup_record.go @@ -9,9 +9,11 @@ const ( ) const ( - // BackupKindFull 全量备份；BackupKindDifferential 差异备份（仅含自基线全量以来的变更）。 + // BackupKindFull 全量归档；BackupKindDifferential 差异归档； + // BackupKindRepository 为可独立恢复的 CDC 内容寻址快照。 BackupKindFull = "full" BackupKindDifferential = "differential" + BackupKindRepository = "repository" ) type BackupRecord struct { @@ -33,7 +35,7 @@ type BackupRecord struct { // Locked 保留锁定（法律保留）：为 true 时该备份不参与保留期/数量自动清理， // 且禁止手动删除，直到显式解锁。用于保护合规快照、迁移前基线等关键备份。 Locked bool `gorm:"column:locked;not null;default:false;index" json:"locked"` - // BackupKind 备份类型：full（全量）/ differential（差异）。 + // BackupKind 备份类型：full（全量）/ differential（差异）/ repository（CDC 快照）。 BackupKind string `gorm:"column:backup_kind;size:16;not null;default:'full';index" json:"backupKind"` // BaseRecordID 差异备份所基于的全量备份记录 ID（全量记录为 0）。 BaseRecordID uint `gorm:"column:base_record_id;index;not null;default:0" json:"baseRecordId"` diff --git a/server/internal/model/backup_task.go b/server/internal/model/backup_task.go index a3409d7..b077f68 100644 --- a/server/internal/model/backup_task.go +++ b/server/internal/model/backup_task.go @@ -12,9 +12,11 @@ const ( ) const ( - // BackupModeFull 全量模式（默认）；BackupModeDifferential 差异模式（仅文件类型本机任务）。 + // BackupModeFull 全量模式（默认）；BackupModeDifferential 差异归档； + // BackupModeRepository 为 CDC 内容寻址仓库模式（仅文件类型本机任务）。 BackupModeFull = "full" BackupModeDifferential = "differential" + BackupModeRepository = "repository" ) const ( @@ -55,7 +57,8 @@ type BackupTask struct { Compression string `gorm:"size:10;not null;default:'gzip'" json:"compression"` Encrypt bool `gorm:"not null;default:false" json:"encrypt"` MaxBackups int `gorm:"column:max_backups;not null;default:10" json:"maxBackups"` - // BackupMode 备份模式：full（全量，默认）/ differential（差异）。差异仅支持本机文件任务。 + // BackupMode 备份模式：full（全量，默认）/ differential（差异归档）/ + // repository（FastCDC 分块、全局去重快照）。后两者仅支持本机文件任务。 BackupMode string `gorm:"column:backup_mode;size:16;not null;default:'full'" json:"backupMode"` // DiffFullIntervalDays 差异模式下强制全量的间隔（天）：最近全量超过该天数则本次自动改为全量， // 限制差异链跨度与单个差异体积。默认 7。 diff --git a/server/internal/service/backup_execution_service.go b/server/internal/service/backup_execution_service.go index 76066b2..ea74e5f 100644 --- a/server/internal/service/backup_execution_service.go +++ b/server/internal/service/backup_execution_service.go @@ -5,11 +5,13 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "hash" "io" "os" "path/filepath" + "sort" "strings" "sync" "time" @@ -62,6 +64,17 @@ type DownloadedArtifact struct { Reader io.ReadCloser } +type temporaryArtifactReader struct { + *os.File + directory string +} + +func (r *temporaryArtifactReader) Close() error { + closeErr := r.File.Close() + removeErr := os.RemoveAll(r.directory) + return errors.Join(closeErr, removeErr) +} + // collectTargetIDs 获取任务关联的所有存储目标 ID func collectTargetIDs(task *model.BackupTask) []uint { if len(task.StorageTargets) > 0 { @@ -102,6 +115,10 @@ type BackupExecutionService struct { bandwidthLimit string // rclone 带宽限制（全局默认，节点配置可覆盖） metrics *metrics.Metrics taskLocks sync.Map + // repositoryLocks serializes immutable index updates per storage target. + // Repository mode is intentionally single-writer in v1 to avoid orphaned + // duplicate packs when two local tasks discover the same missing chunk. + repositoryLocks sync.Map } // SetMetrics 注入 Prometheus 采集器。nil 时所有埋点退化为 no-op。 @@ -211,6 +228,25 @@ func (s *BackupExecutionService) DownloadRecord(ctx context.Context, recordID ui if err != nil { return nil, err } + if record.BackupKind == model.BackupKindRepository { + tempDir, err := os.MkdirTemp(s.tempDir, "repository-download-*") + if err != nil { + return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法创建 CDC 导出目录", err) + } + exportName := fmt.Sprintf("backupx-record-%d.tar", record.ID) + exportPath := filepath.Join(tempDir, exportName) + store := backup.NewRepositoryStore(s.cipher.Key()) + if err := store.ExportTar(ctx, provider, record.StoragePath, exportPath); err != nil { + cleanupErr := os.RemoveAll(tempDir) + return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法从 CDC 仓库导出归档", errors.Join(err, cleanupErr)) + } + file, err := os.Open(exportPath) + if err != nil { + cleanupErr := os.RemoveAll(tempDir) + return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法打开 CDC 导出归档", errors.Join(err, cleanupErr)) + } + return &DownloadedArtifact{FileName: exportName, Reader: &temporaryArtifactReader{File: file, directory: tempDir}}, nil + } reader, err := provider.Download(ctx, record.StoragePath) if err != nil { return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法下载备份文件", err) @@ -234,6 +270,16 @@ func (s *BackupExecutionService) RestoreRecord(ctx context.Context, recordID uin if task == nil { return apperror.New(404, "BACKUP_TASK_NOT_FOUND", "关联的备份任务不存在，无法执行恢复", fmt.Errorf("backup task %d not found", record.TaskID)) } + if record.BackupKind == model.BackupKindRepository { + spec, specErr := s.buildTaskSpec(task, record.StartedAt) + if specErr != nil { + return specErr + } + if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, spec, backup.NopLogWriter{}); err != nil { + return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "从 CDC 仓库恢复备份失败", err) + } + return nil + } tempDir, err := os.MkdirTemp("", "backupx-restore-*") if err != nil { return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法创建恢复目录", err) @@ -291,6 +337,58 @@ func (s *BackupExecutionService) DeleteRecord(ctx context.Context, recordID uint fmt.Sprintf("该全量备份仍有 %d 个差异备份依赖它，删除会导致这些差异无法恢复。请先删除相关差异备份或等待其过期。", deps), nil) } } + if record.BackupKind == model.BackupKindRepository { + copies := []StorageUploadResultItem{{ + StorageTargetID: record.StorageTargetID, + Status: model.BackupRecordStatusSuccess, + StoragePath: record.StoragePath, + }} + if strings.TrimSpace(record.StorageUploadResults) != "" { + if err := json.Unmarshal([]byte(record.StorageUploadResults), &copies); err != nil { + return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法解析 CDC 仓库副本信息，已停止删除以避免遗留数据", err) + } + } + copyPaths := make(map[uint]string, len(copies)) + for _, copy := range copies { + if strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) && strings.TrimSpace(copy.StoragePath) != "" { + copyPaths[copy.StorageTargetID] = copy.StoragePath + } + } + targetIDs := make([]uint, 0, len(copyPaths)) + for targetID := range copyPaths { + targetIDs = append(targetIDs, targetID) + } + sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] }) + unlocks := make([]func(), 0, len(targetIDs)) + for _, targetID := range targetIDs { + unlocks = append(unlocks, s.acquireRepositoryLock(targetID)) + } + defer func() { + for index := len(unlocks) - 1; index >= 0; index-- { + unlocks[index]() + } + }() + providers := make(map[uint]storage.StorageProvider, len(targetIDs)) + for _, targetID := range targetIDs { + provider, resolveErr := s.resolveProvider(ctx, targetID) + if resolveErr != nil { + return resolveErr + } + if deleteErr := provider.Delete(ctx, copyPaths[targetID]); deleteErr != nil { + return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除 CDC 仓库快照", deleteErr) + } + providers[targetID] = provider + } + for targetID, provider := range providers { + if _, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider); pruneErr != nil { + return apperror.Internal("BACKUP_REPOSITORY_PRUNE_FAILED", fmt.Sprintf("无法清理存储目标 %d 的 CDC 仓库；记录暂时保留以便重试", targetID), pruneErr) + } + } + if err := s.records.Delete(ctx, recordID); err != nil { + return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份记录", err) + } + return nil + } if remote, err := s.deleteRemoteLocalDiskObject(ctx, record); err != nil { return err } else if !remote && strings.TrimSpace(record.StoragePath) != "" { @@ -384,6 +482,9 @@ func (s *BackupExecutionService) startTask(ctx context.Context, id uint, async b return nil, perr } } + if strings.EqualFold(task.BackupMode, model.BackupModeRepository) && s.resolveRemoteNode(ctx, resolvedNodeID) != nil { + return nil, apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前仅支持 Master 本机单写者执行", nil) + } startedAt := s.now() // 取第一个存储目标 ID 做兼容 primaryTargetID := task.StorageTargetID @@ -630,6 +731,141 @@ func (s *BackupExecutionService) resolveDifferentialBase(ctx context.Context, ta return 0, backup.Manifest{}, false } +type repositoryTaskResult struct { + fileName string + logicalSize int64 + checksum string + storagePath string + storageTargetID uint + manifestJSON string + uploadResults []StorageUploadResultItem + providers map[uint]storage.StorageProvider +} + +func (s *BackupExecutionService) executeRepositoryTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time, spec backup.TaskSpec, logger *backup.ExecutionLogger) (*repositoryTaskResult, error) { + store := backup.NewRepositoryStore(s.cipher.Key()) + plan, err := store.BuildPlan(ctx, spec, logger) + if err != nil { + return nil, err + } + defer func() { + if closeErr := plan.Close(); closeErr != nil { + logger.Warnf("清理 CDC 临时计划失败：%v", closeErr) + } + }() + manifestBytes, err := backup.EncodeManifest(plan.Manifest) + if err != nil { + return nil, fmt.Errorf("encode repository manifest: %w", err) + } + targetIDs := collectTargetIDs(task) + if len(targetIDs) == 0 { + return nil, fmt.Errorf("没有关联的存储目标") + } + storageUsage, usageErr := s.storageUsageSnapshot(ctx) + if usageErr != nil { + logger.Warnf("读取存储目标用量失败，跳过本次软配额校验：%v", usageErr) + storageUsage = map[uint]int64{} + } + + snapshotKey := store.SnapshotKey(task.ID, recordID, startedAt) + result := &repositoryTaskResult{ + fileName: filepath.Base(snapshotKey), + logicalSize: plan.LogicalSize, + storagePath: snapshotKey, + manifestJSON: string(manifestBytes), + uploadResults: make([]StorageUploadResultItem, 0, len(targetIDs)), + providers: make(map[uint]storage.StorageProvider), + } + var failures []string + for _, targetID := range targetIDs { + target, findErr := s.targets.FindByID(ctx, targetID) + targetName := fmt.Sprintf("target-%d", targetID) + if findErr == nil && target != nil { + targetName = target.Name + } + if findErr != nil || target == nil { + message := "存储目标不存在" + if findErr != nil { + message = findErr.Error() + } + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message}) + failures = append(failures, fmt.Sprintf("%s: %s", targetName, message)) + continue + } + provider, resolveErr := s.resolveProviderForNode(ctx, targetID, task.NodeID) + if resolveErr != nil { + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: resolveErr.Error()}) + failures = append(failures, fmt.Sprintf("%s: %v", targetName, resolveErr)) + continue + } + logger.Infof("同步 CDC 仓库到存储目标：%s", targetName) + unlock := s.acquireRepositoryLock(targetID) + estimatedSize, estimateErr := store.EstimateUploadSize(ctx, provider, plan) + if estimateErr != nil { + unlock() + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: estimateErr.Error()}) + failures = append(failures, fmt.Sprintf("%s: %v", targetName, estimateErr)) + continue + } + if target.QuotaBytes > 0 && storageUsage[targetID]+estimatedSize > target.QuotaBytes { + unlock() + message := fmt.Sprintf("超出存储目标配额（%d + 预计 %d > %d）", storageUsage[targetID], estimatedSize, target.QuotaBytes) + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: message}) + failures = append(failures, fmt.Sprintf("%s: %s", targetName, message)) + continue + } + upload, uploadErr := store.Upload(ctx, provider, plan, snapshotKey) + unlock() + if uploadErr != nil { + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: uploadErr.Error()}) + failures = append(failures, fmt.Sprintf("%s: %v", targetName, uploadErr)) + logger.Warnf("存储目标 %s CDC 仓库同步失败：%v", targetName, uploadErr) + continue + } + result.uploadResults = append(result.uploadResults, StorageUploadResultItem{ + StorageTargetID: targetID, StorageTargetName: targetName, Status: "success", + StoragePath: upload.SnapshotKey, FileSize: upload.UploadedBytes, + }) + result.providers[targetID] = provider + if result.storageTargetID == 0 { + result.storageTargetID = targetID + result.checksum = upload.Checksum + } + logger.Infof("存储目标 %s CDC 同步完成：新块 %d/%d，复用 %d bytes，实际上传 %d bytes", targetName, upload.NewChunks, upload.UniqueChunks, upload.ReusedBytes, upload.UploadedBytes) + } + if result.storageTargetID == 0 { + return nil, fmt.Errorf("所有存储目标 CDC 仓库同步均失败：%s", strings.Join(failures, "; ")) + } + if len(failures) > 0 { + logger.Warnf("部分存储目标 CDC 仓库同步失败：%s", strings.Join(failures, "; ")) + } + if s.dependentsResolver != nil { + go func(upstreamID uint, upstreamName string) { + dependents, resolveErr := s.dependentsResolver.TriggerDependents(context.Background(), upstreamID) + if resolveErr != nil { + logger.Warnf("解析任务 %s 的下游依赖失败：%v", upstreamName, resolveErr) + return + } + for _, dependentID := range dependents { + if _, runErr := s.RunTaskByID(context.Background(), dependentID); runErr != nil { + logger.Warnf("触发下游任务 #%d 失败（上游: %s）：%v", dependentID, upstreamName, runErr) + } else { + logger.Infof("已触发下游任务 #%d（上游: %s）", dependentID, upstreamName) + } + } + }(task.ID, task.Name) + } + return result, nil +} + +func (s *BackupExecutionService) acquireRepositoryLock(targetID uint) func() { + created := &sync.Mutex{} + actual, _ := s.repositoryLocks.LoadOrStore(targetID, created) + lock := actual.(*sync.Mutex) + lock.Lock() + return lock.Unlock +} + func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time) { // 节点级并发限流：当任务绑定节点且节点配置了 MaxConcurrent>0， // 该节点上所有任务共享一个节点专属 semaphore，互相排队 @@ -658,18 +894,33 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba backupKind := model.BackupKindFull var baseRecordID uint var manifestJSON string + var repositoryProviders map[uint]storage.StorageProvider completeRecord := func() { + readyForRepositoryRetention := status == model.BackupRecordStatusSuccess if finalizeErr := s.finalizeRecord(ctx, task, recordID, startedAt, status, errMessage, logger.String(), fileName, fileSize, checksum, storagePath, selectedStorageTargetID); finalizeErr != nil { logger.Errorf("写回备份记录失败：%v", finalizeErr) + readyForRepositoryRetention = false } // 采集任务执行结果到 Prometheus（耗时 + 产出字节 + 状态计数） s.metrics.ObserveTaskRun(task.Type, status, time.Since(startedAt).Seconds(), fileSize) // 写入多目标上传结果 if len(uploadResults) > 0 { - if resultsJSON, marshalErr := json.Marshal(uploadResults); marshalErr == nil { - if record, findErr := s.records.FindByID(ctx, recordID); findErr == nil && record != nil { - record.StorageUploadResults = string(resultsJSON) - _ = s.records.Update(ctx, record) + resultsJSON, marshalErr := json.Marshal(uploadResults) + if marshalErr != nil { + logger.Warnf("序列化多目标上传结果失败：%v", marshalErr) + readyForRepositoryRetention = false + } else if record, findErr := s.records.FindByID(ctx, recordID); findErr != nil || record == nil { + if findErr != nil { + logger.Warnf("读取备份记录以写回多目标结果失败：%v", findErr) + } else { + logger.Warnf("备份记录 #%d 不存在，无法写回多目标结果", recordID) + } + readyForRepositoryRetention = false + } else { + record.StorageUploadResults = string(resultsJSON) + if updateErr := s.records.Update(ctx, record); updateErr != nil { + logger.Warnf("写回多目标上传结果失败：%v", updateErr) + readyForRepositoryRetention = false } } } @@ -681,6 +932,36 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba record.Manifest = manifestJSON if updErr := s.records.Update(ctx, record); updErr != nil { logger.Warnf("写回差异链信息失败：%v", updErr) + readyForRepositoryRetention = false + } + } else { + if findErr != nil { + logger.Warnf("读取备份记录以写回备份类型失败：%v", findErr) + } else { + logger.Warnf("备份记录 #%d 不存在，无法写回备份类型", recordID) + } + readyForRepositoryRetention = false + } + } + if readyForRepositoryRetention && backupKind == model.BackupKindRepository && s.retention != nil && len(repositoryProviders) > 0 { + targetIDs := make([]uint, 0, len(repositoryProviders)) + for targetID := range repositoryProviders { + targetIDs = append(targetIDs, targetID) + } + sort.Slice(targetIDs, func(i, j int) bool { return targetIDs[i] < targetIDs[j] }) + unlocks := make([]func(), 0, len(targetIDs)) + for _, targetID := range targetIDs { + unlocks = append(unlocks, s.acquireRepositoryLock(targetID)) + } + cleanupResult, cleanupErr := s.retention.CleanupProviders(ctx, task, repositoryProviders) + for index := len(unlocks) - 1; index >= 0; index-- { + unlocks[index]() + } + if cleanupErr != nil { + logger.Warnf("执行 CDC 仓库保留策略失败：%v", cleanupErr) + } else { + for _, warning := range cleanupResult.Warnings { + logger.Warnf("CDC 仓库保留策略警告：%s", warning) } } } @@ -701,6 +982,26 @@ func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.Ba logger.Errorf("构建任务运行时配置失败：%v", err) return } + if task.Type == model.BackupTaskTypeFile && strings.EqualFold(task.BackupMode, model.BackupModeRepository) { + backupKind = model.BackupKindRepository + repositoryResult, repositoryErr := s.executeRepositoryTask(ctx, task, recordID, startedAt, spec, logger) + if repositoryErr != nil { + errMessage = repositoryErr.Error() + logger.Errorf("执行 CDC 仓库备份失败：%v", repositoryErr) + return + } + fileName = repositoryResult.fileName + fileSize = repositoryResult.logicalSize + checksum = repositoryResult.checksum + storagePath = repositoryResult.storagePath + selectedStorageTargetID = repositoryResult.storageTargetID + uploadResults = repositoryResult.uploadResults + repositoryProviders = repositoryResult.providers + manifestJSON = repositoryResult.manifestJSON + status = model.BackupRecordStatusSuccess + logger.Infof("CDC 仓库备份执行完成") + return + } // 差异备份：解析基线全量，命中则切换为差异模式（仅本机文件任务）。 if baseID, baseManifest, ok := s.resolveDifferentialBase(ctx, task); ok { spec.Differential = true diff --git a/server/internal/service/backup_execution_service_test.go b/server/internal/service/backup_execution_service_test.go index 5f9ec15..abbcbd0 100644 --- a/server/internal/service/backup_execution_service_test.go +++ b/server/internal/service/backup_execution_service_test.go @@ -1,6 +1,7 @@ package service import ( + "bytes" "context" "errors" "fmt" @@ -155,6 +156,108 @@ func TestBackupExecutionServiceRunTaskByIDSync(t *testing.T) { } } +func TestBackupExecutionServiceRepositoryModeRoundTrip(t *testing.T) { + executionService, recordService, tasks, _, records, sourceDir, storageDir := newExecutionTestServices(t) + ctx := context.Background() + task, err := tasks.FindByID(ctx, 1) + if err != nil { + t.Fatalf("FindByID task returned error: %v", err) + } + task.BackupMode = model.BackupModeRepository + task.Compression = "zstd" + if err := tasks.Update(ctx, task); err != nil { + t.Fatalf("Update repository task returned error: %v", err) + } + large := make([]byte, 4<<20) + for index := range large { + large[index] = byte((index * 31) % 251) + } + largePath := filepath.Join(sourceDir, "large.bin") + if err := os.WriteFile(largePath, large, 0o640); err != nil { + t.Fatalf("write large fixture: %v", err) + } + + first, err := executionService.RunTaskByIDSync(ctx, task.ID) + if err != nil { + t.Fatalf("first repository backup returned error: %v", err) + } + if first.Status != model.BackupRecordStatusSuccess || first.BackupKind != model.BackupKindRepository { + t.Fatalf("unexpected first repository record: %#v", first) + } + if !strings.HasPrefix(first.StoragePath, ".backupx/repository/v1/snapshots/") { + t.Fatalf("unexpected repository snapshot path: %s", first.StoragePath) + } + + large[2<<20] ^= 0xff + if err := os.WriteFile(largePath, large, 0o640); err != nil { + t.Fatalf("modify large fixture: %v", err) + } + second, err := executionService.RunTaskByIDSync(ctx, task.ID) + if err != nil { + t.Fatalf("second repository backup returned error: %v", err) + } + stored, err := records.FindByID(ctx, second.ID) + if err != nil { + t.Fatalf("FindByID repository record returned error: %v", err) + } + if stored == nil || stored.BackupKind != model.BackupKindRepository || stored.Manifest == "" { + t.Fatalf("repository metadata was not persisted: %#v", stored) + } + + download, err := recordService.Download(ctx, second.ID) + if err != nil { + t.Fatalf("export repository snapshot returned error: %v", err) + } + exported, readErr := io.ReadAll(download.Reader) + closeErr := download.Reader.Close() + if readErr != nil || closeErr != nil { + t.Fatalf("read repository export: read=%v close=%v", readErr, closeErr) + } + if len(exported) == 0 || !strings.HasSuffix(download.FileName, ".tar") { + t.Fatalf("unexpected repository export: name=%s size=%d", download.FileName, len(exported)) + } + + if err := os.WriteFile(largePath, bytes.Repeat([]byte{0}, len(large)), 0o640); err != nil { + t.Fatalf("damage source before restore: %v", err) + } + if err := executionService.RestoreRecord(ctx, second.ID); err != nil { + t.Fatalf("restore repository record returned error: %v", err) + } + restored, err := os.ReadFile(largePath) + if err != nil { + t.Fatalf("read restored source: %v", err) + } + if !bytes.Equal(restored, large) { + t.Fatalf("repository restore did not reproduce the source") + } + + if err := recordService.Delete(ctx, first.ID); err != nil { + t.Fatalf("delete first repository record: %v", err) + } + if err := recordService.Delete(ctx, second.ID); err != nil { + t.Fatalf("delete second repository record: %v", err) + } + packRoot := filepath.Join(storageDir, filepath.FromSlash(".backupx/repository/v1/packs")) + remainingPacks := 0 + if err := filepath.Walk(packRoot, func(_ string, info os.FileInfo, walkErr error) error { + if walkErr != nil { + if os.IsNotExist(walkErr) { + return nil + } + return walkErr + } + if info != nil && !info.IsDir() { + remainingPacks++ + } + return nil + }); err != nil { + t.Fatalf("inspect repository packs: %v", err) + } + if remainingPacks != 0 { + t.Fatalf("repository prune left %d packs after deleting all snapshots", remainingPacks) + } +} + func TestBackupExecutionServiceNodePoolSelectionDoesNotPersistTaskNodeID(t *testing.T) { executionService, _, tasks, _, records, _, _ := newExecutionTestServices(t) ctx := context.Background() diff --git a/server/internal/service/backup_task_service.go b/server/internal/service/backup_task_service.go index 998f327..87eee42 100644 --- a/server/internal/service/backup_task_service.go +++ b/server/internal/service/backup_task_service.go @@ -57,8 +57,8 @@ type BackupTaskUpsertInput struct { KeepWeekly int `json:"keepWeekly"` KeepMonthly int `json:"keepMonthly"` KeepYearly int `json:"keepYearly"` - // BackupMode 备份模式：full（默认）/ differential（差异，仅文件类型本机任务） - BackupMode string `json:"backupMode" binding:"omitempty,oneof=full differential"` + // BackupMode 备份模式：full（默认）/ differential（差异归档）/ repository（CDC 去重仓库） + BackupMode string `json:"backupMode" binding:"omitempty,oneof=full differential repository"` DiffFullIntervalDays int `json:"diffFullIntervalDays"` // 备份复制目标存储 ID 列表（3-2-1 规则） ReplicationTargetIDs []uint `json:"replicationTargetIds"` @@ -414,21 +414,50 @@ func (s *BackupTaskService) cleanupRemoteFiles(ctx context.Context, taskID uint) recordCount = len(records) // 缓存 provider 避免同一存储目标重复创建连接 providerCache := make(map[uint]storage.StorageProvider) + repositoryProviders := make(map[uint]storage.StorageProvider) for _, record := range records { - if strings.TrimSpace(record.StoragePath) == "" { - continue + copies := []StorageUploadResultItem{{ + StorageTargetID: record.StorageTargetID, + Status: model.BackupRecordStatusSuccess, + StoragePath: record.StoragePath, + }} + if strings.TrimSpace(record.StorageUploadResults) != "" { + var storedCopies []StorageUploadResultItem + if unmarshalErr := json.Unmarshal([]byte(record.StorageUploadResults), &storedCopies); unmarshalErr == nil { + copies = storedCopies + } } - provider, ok := providerCache[record.StorageTargetID] - if !ok { - provider, err = s.resolveStorageProvider(ctx, record.StorageTargetID) - if err != nil { + seenTargets := make(map[uint]struct{}, len(copies)) + for _, copy := range copies { + if !strings.EqualFold(copy.Status, model.BackupRecordStatusSuccess) || strings.TrimSpace(copy.StoragePath) == "" { continue } - providerCache[record.StorageTargetID] = provider + if _, seen := seenTargets[copy.StorageTargetID]; seen { + continue + } + seenTargets[copy.StorageTargetID] = struct{}{} + provider, ok := providerCache[copy.StorageTargetID] + if !ok { + provider, err = s.resolveStorageProvider(ctx, copy.StorageTargetID) + if err != nil { + continue + } + providerCache[copy.StorageTargetID] = provider + } + if err := provider.Delete(ctx, copy.StoragePath); err == nil { + cleanedFiles++ + if record.BackupKind == model.BackupKindRepository { + repositoryProviders[copy.StorageTargetID] = provider + } + } } - if err := provider.Delete(ctx, record.StoragePath); err == nil { - cleanedFiles++ + } + for _, provider := range repositoryProviders { + pruned, pruneErr := backup.NewRepositoryStore(s.cipher.Key()).Prune(ctx, provider) + if pruneErr != nil { + continue } + cleanedFiles += pruned.DeletedIndexes + pruned.DeletedPacks } return recordCount, cleanedFiles } @@ -530,6 +559,17 @@ func (s *BackupTaskService) validateInput(ctx context.Context, existing *model.B return apperror.BadRequest("BACKUP_TASK_DIFF_REMOTE_UNSUPPORTED", "差异备份当前仅支持本机 Master 执行，请将任务固定在本机或改用全量备份。", nil) } } + if strings.EqualFold(strings.TrimSpace(input.BackupMode), model.BackupModeRepository) { + if input.Type != model.BackupTaskTypeFile { + return apperror.BadRequest("BACKUP_TASK_REPOSITORY_UNSUPPORTED", "CDC 仓库模式仅支持文件目录类型任务", nil) + } + if strings.TrimSpace(input.NodePoolTag) != "" || (fixedNode != nil && !fixedNode.IsLocal) { + return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REMOTE_UNSUPPORTED", "CDC 仓库模式当前采用单写者索引，仅支持 Master 本机执行。远程服务器备份请暂用全量模式。", nil) + } + if len(input.ReplicationTargetIDs) > 0 { + return apperror.BadRequest("BACKUP_TASK_REPOSITORY_REPLICATION_UNSUPPORTED", "CDC 仓库快照不能使用对象级复制；请直接为任务选择多个存储目标以生成完整仓库副本。", nil) + } + } if input.RetentionDays < 0 { return apperror.BadRequest("BACKUP_TASK_INVALID", "保留天数不能小于 0", nil) } @@ -935,10 +975,17 @@ func decodeExtraConfig(value string) (map[string]any, error) { return result, nil } -// normalizeBackupMode 归一化备份模式：仅文件类型可启用差异，其余一律全量（双保险，防绕过校验）。 +// normalizeBackupMode 归一化备份模式：仅文件类型可启用差异或 CDC 仓库， +// 其余一律全量（双保险，防绕过校验）。 func normalizeBackupMode(mode, taskType string) string { - if strings.EqualFold(strings.TrimSpace(mode), model.BackupModeDifferential) && normalizeBackupTaskType(taskType) == model.BackupTaskTypeFile { + if normalizeBackupTaskType(taskType) != model.BackupTaskTypeFile { + return model.BackupModeFull + } + switch strings.ToLower(strings.TrimSpace(mode)) { + case model.BackupModeDifferential: return model.BackupModeDifferential + case model.BackupModeRepository: + return model.BackupModeRepository } return model.BackupModeFull } diff --git a/server/internal/service/restore_service.go b/server/internal/service/restore_service.go index ba1c0db..edf22e0 100644 --- a/server/internal/service/restore_service.go +++ b/server/internal/service/restore_service.go @@ -321,6 +321,13 @@ func (s *RestoreService) restoreArtifact(ctx context.Context, record *model.Back if err != nil { return fmt.Errorf("创建存储客户端失败：%w", err) } + if record.BackupKind == model.BackupKindRepository { + logger.Infof("读取 CDC 仓库快照：%s", record.StoragePath) + if err := backup.NewRepositoryStore(s.cipher.Key()).Restore(ctx, provider, record.StoragePath, spec, logger); err != nil { + return fmt.Errorf("恢复 CDC 仓库快照失败：%w", err) + } + return nil + } recDir, err := os.MkdirTemp(parentTempDir, fmt.Sprintf("rec-%d-*", record.ID)) if err != nil { return fmt.Errorf("创建恢复子目录失败：%w", err) @@ -368,6 +375,9 @@ func (s *RestoreService) buildRestoreChain(ctx context.Context, record *model.Ba } func backupKindLabel(kind string) string { + if kind == model.BackupKindRepository { + return "CDC 仓库快照" + } if kind == model.BackupKindDifferential { return "差异" } diff --git a/server/internal/service/verification_service.go b/server/internal/service/verification_service.go index 4350d8c..45e7f40 100644 --- a/server/internal/service/verification_service.go +++ b/server/internal/service/verification_service.go @@ -299,6 +299,20 @@ func (s *VerificationService) executeLocally(ctx context.Context, verID uint, ta logger.Errorf("创建存储客户端失败：%v", err) return } + if backupRecord.BackupKind == model.BackupKindRepository { + logger.Infof("验证 CDC 仓库快照及全部引用块：%s", backupRecord.StoragePath) + report, verifyErr := backup.NewRepositoryStore(s.cipher.Key()).Verify(ctx, provider, backupRecord.StoragePath, backupRecord.Checksum) + if verifyErr != nil { + errMessage = verifyErr.Error() + summary = "CDC 仓库完整性校验失败" + logger.Errorf("验证未通过：%v", verifyErr) + return + } + status = model.VerificationRecordStatusSuccess + summary = fmt.Sprintf("CDC 仓库完整性校验通过：%d 个条目、%d 个唯一块、%d bytes", report.Entries, report.Chunks, report.Bytes) + logger.Infof("%s", summary) + return + } fileName := backupRecord.FileName if strings.TrimSpace(fileName) == "" { fileName = filepath.Base(backupRecord.StoragePath) diff --git a/server/internal/storage/rclone/provider.go b/server/internal/storage/rclone/provider.go index 60f1257..ff33147 100644 --- a/server/internal/storage/rclone/provider.go +++ b/server/internal/storage/rclone/provider.go @@ -2,6 +2,7 @@ package rclone import ( "context" + "errors" "fmt" "io" "sort" @@ -68,13 +69,56 @@ func (p *Provider) Download(ctx context.Context, objectKey string) (io.ReadClose return reader, nil } +// DownloadRange reads one slice from an object. Most object-storage backends +// map this to a native HTTP Range request. Backends that reject ranged reads +// fall back to a full stream while preserving the same interface contract. +func (p *Provider) DownloadRange(ctx context.Context, objectKey string, offset, length int64) (io.ReadCloser, error) { + if offset < 0 || length <= 0 { + return nil, fmt.Errorf("rclone download range %s: invalid offset=%d length=%d", objectKey, offset, length) + } + obj, err := p.rfs.NewObject(ctx, objectKey) + if err != nil { + return nil, fmt.Errorf("rclone find object %s: %w", objectKey, err) + } + reader, rangeErr := obj.Open(ctx, &fs.RangeOption{Start: offset, End: offset + length - 1}) + if rangeErr == nil { + return reader, nil + } + reader, err = obj.Open(ctx) + if err != nil { + return nil, fmt.Errorf("rclone download range %s (range: %v; fallback: %w)", objectKey, rangeErr, err) + } + if offset > 0 { + if _, err := io.CopyN(io.Discard, reader, offset); err != nil { + closeErr := reader.Close() + return nil, errors.Join(fmt.Errorf("rclone seek object %s: %w", objectKey, err), closeErr) + } + } + return &limitedReadCloser{Reader: io.LimitReader(reader, length), closer: reader}, nil +} + +type limitedReadCloser struct { + io.Reader + closer io.Closer +} + +func (r *limitedReadCloser) Close() error { + return r.closer.Close() +} + // Delete 通过 rclone 删除远端对象。 func (p *Provider) Delete(ctx context.Context, objectKey string) error { obj, err := p.rfs.NewObject(ctx, objectKey) if err != nil { + if errors.Is(err, fs.ErrorObjectNotFound) || errors.Is(err, fs.ErrorDirNotFound) { + return nil + } return fmt.Errorf("rclone find object %s: %w", objectKey, err) } if err := obj.Remove(ctx); err != nil { + if errors.Is(err, fs.ErrorObjectNotFound) || errors.Is(err, fs.ErrorDirNotFound) { + return nil + } return fmt.Errorf("rclone delete %s: %w", objectKey, err) } return nil @@ -102,6 +146,9 @@ func (p *Provider) List(ctx context.Context, prefix string) ([]storage.ObjectInf return nil }) if err != nil { + if errors.Is(err, fs.ErrorDirNotFound) || errors.Is(err, fs.ErrorObjectNotFound) { + return []storage.ObjectInfo{}, nil + } return nil, fmt.Errorf("rclone list %s: %w", prefix, err) } return items, nil diff --git a/server/internal/storage/types.go b/server/internal/storage/types.go index 92672e7..3abc664 100644 --- a/server/internal/storage/types.go +++ b/server/internal/storage/types.go @@ -49,6 +49,13 @@ type StorageProvider interface { List(ctx context.Context, prefix string) ([]ObjectInfo, error) } +// StorageRangeDownloader is an optional capability used by packed repository +// backups. Implementations return exactly the requested byte range when the +// backend supports ranged reads and may transparently fall back to a full read. +type StorageRangeDownloader interface { + DownloadRange(ctx context.Context, objectKey string, offset, length int64) (io.ReadCloser, error) +} + type ProviderFactory interface { Type() ProviderType } @@ -151,4 +158,3 @@ type FTPConfig struct { type StorageDirCleaner interface { RemoveEmptyDirs(ctx context.Context, prefix string) error } - diff --git a/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx b/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx index 7a24079..f4722cd 100644 --- a/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx +++ b/web/src/components/backup-tasks/BackupTaskFormDrawer.tsx @@ -251,6 +251,12 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa if (validPaths.length === 0 && !value.sourcePath.trim()) { return '请输入至少一个源路径' } + if (value.backupMode === 'repository' && (((value.nodeId ?? 0) > 0 && value.nodeId !== localNodeId) || value.nodePoolTag?.trim())) { + return 'CDC 仓库模式当前仅支持 Master 本机执行' + } + if (value.backupMode === 'repository' && value.replicationTargetIds.length > 0) { + return 'CDC 仓库模式请直接多选存储目标，不能使用对象级副本复制' + } } if (isSQLiteBackupTask(value.type) && !value.dbPath.trim()) { return '请输入 SQLite 数据库路径' @@ -314,7 +320,13 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa onChange={(value) => { const nodeId = Number(value ?? 0) // 固定节点与节点池互斥：切到固定节点时清空 NodePoolTag - updateDraft(nodeId > 0 ? { nodeId, nodePoolTag: '' } : { nodeId }) + updateDraft(nodeId > 0 + ? { + nodeId, + nodePoolTag: '', + backupMode: nodeId !== localNodeId && draft.backupMode === 'repository' ? 'full' : draft.backupMode, + } + : { nodeId }) }} /> @@ -327,7 +339,10 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa placeholder="填写标签后从节点池动态调度（与固定节点互斥）" value={draft.nodePoolTag ?? ''} disabled={(draft.nodeId ?? 0) > 0} - onChange={(value) => updateDraft({ nodePoolTag: value })} + onChange={(value) => updateDraft({ + nodePoolTag: value, + backupMode: value.trim() && draft.backupMode === 'repository' ? 'full' : draft.backupMode, + })} /> 执行节点选"本机 / 未指定"时可启用；从节点 Labels 命中此 tag 的在线节点中按当前运行任务数最少的挑选一台执行。 @@ -600,8 +615,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa options={[ { label: '全量备份', value: 'full' }, { label: '差异备份（仅文件、本机）', value: 'differential' }, + { label: 'CDC 去重仓库（仅文件、本机）', value: 'repository' }, ]} - onChange={(value) => updateDraft({ backupMode: value as BackupMode })} + onChange={(value) => updateDraft(value === 'repository' + ? { backupMode: value as BackupMode, nodeId: 0, nodePoolTag: '', replicationTargetIds: [] } + : { backupMode: value as BackupMode })} /> {draft.backupMode === 'differential' && (
@@ -618,6 +636,11 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa />
)} + {draft.backupMode === 'repository' && ( + + 文件按内容边界切块并写入全局分块池；相同数据跨文件、跨快照只上传一次。新块会合并为 pack，恢复时通过索引按需读取。当前版本采用单写者索引，因此固定在 Master 本机执行；需要多副本时请直接多选上方存储目标。 + + )} )}
@@ -736,10 +759,13 @@ export function BackupTaskFormDrawer({ visible, loading, initialValue, storageTa value={draft.replicationTargetIds} placeholder="选择副本目标（不选 = 不启用复制）" options={storageTargetOptions.filter((opt) => !(draft.storageTargetIds ?? []).includes(opt.value as number))} + disabled={draft.backupMode === 'repository'} onChange={(values: number[]) => updateDraft({ replicationTargetIds: values })} /> - 备份成功后自动镜像到副本存储。满足 3-2-1 规则：至少 2 份副本、至少 1 份异地。建议选不同 provider 的目标。 + {draft.backupMode === 'repository' + ? 'CDC 仓库包含共享 pack 与索引，不能只复制单个快照对象；请在“存储目标”中直接多选以生成完整仓库副本。' + : '备份成功后自动镜像到副本存储。满足 3-2-1 规则：至少 2 份副本、至少 1 份异地。建议选不同 provider 的目标。'}
diff --git a/web/src/pages/backup-records/BackupRecordsPage.tsx b/web/src/pages/backup-records/BackupRecordsPage.tsx index 3040f7e..a5913fb 100644 --- a/web/src/pages/backup-records/BackupRecordsPage.tsx +++ b/web/src/pages/backup-records/BackupRecordsPage.tsx @@ -114,6 +114,7 @@ export function BackupRecordsPage() { {record.fileName || '-'} {record.locked && 已锁定} {record.backupKind === 'differential' && 差异} + {record.backupKind === 'repository' && CDC} {formatBytes(record.fileSize)} {record.checksum && ( diff --git a/web/src/types/backup-records.ts b/web/src/types/backup-records.ts index 01c980f..0da3d32 100644 --- a/web/src/types/backup-records.ts +++ b/web/src/types/backup-records.ts @@ -26,7 +26,7 @@ export interface BackupRecordSummary { startedAt: string completedAt?: string locked: boolean - backupKind: 'full' | 'differential' + backupKind: 'full' | 'differential' | 'repository' } export interface BackupRecordContentEntry { diff --git a/web/src/types/backup-tasks.ts b/web/src/types/backup-tasks.ts index b254ad1..eeeeb20 100644 --- a/web/src/types/backup-tasks.ts +++ b/web/src/types/backup-tasks.ts @@ -1,7 +1,7 @@ export type BackupTaskType = 'file' | 'mysql' | 'sqlite' | 'postgresql' | 'saphana' | 'mongodb' export type BackupTaskStatus = 'idle' | 'running' | 'success' | 'failed' export type BackupCompression = 'gzip' | 'zstd' | 'none' -export type BackupMode = 'full' | 'differential' +export type BackupMode = 'full' | 'differential' | 'repository' export interface BackupTaskSummary { id: number