From 1e05fa0dd702003b09de65187bb62937aad666e9 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 7 Aug 2026 16:53:23 +0000 Subject: [PATCH] Apply linter autofixes: internal/snapshot (refs #61) --- internal/snapshot/backup_test.go | 41 +++- internal/snapshot/exclude_test.go | 34 +++ internal/snapshot/file_change_test.go | 15 +- internal/snapshot/manifest.go | 2 + internal/snapshot/permission_error_test.go | 6 +- internal/snapshot/progress.go | 18 +- internal/snapshot/remotekey.go | 1 + internal/snapshot/scanner.go | 232 +++++++++++++++++---- internal/snapshot/scanner_test.go | 27 ++- internal/snapshot/snapshot.go | 160 +++++++++++--- internal/snapshot/snapshot_test.go | 13 +- 11 files changed, 452 insertions(+), 97 deletions(-) diff --git a/internal/snapshot/backup_test.go b/internal/snapshot/backup_test.go index fa6b9b4..b310fe7 100644 --- a/internal/snapshot/backup_test.go +++ b/internal/snapshot/backup_test.go @@ -4,6 +4,8 @@ import ( "context" "crypto/sha256" "database/sql" + "encoding/hex" + "errors" "fmt" "io" "io/fs" @@ -30,6 +32,7 @@ func NewMockS3Client() *MockS3Client { func (m *MockS3Client) PutBlob(ctx context.Context, hash string, data []byte) error { m.storage[hash] = data + return nil } @@ -38,11 +41,13 @@ func (m *MockS3Client) GetBlob(ctx context.Context, hash string) ([]byte, error) if !ok { return nil, fmt.Errorf("blob not found: %s", hash) } + return data, nil } func (m *MockS3Client) BlobExists(ctx context.Context, hash string) (bool, error) { _, ok := m.storage[hash] + return ok, nil } @@ -81,12 +86,15 @@ func TestBackupWithInMemoryFS(t *testing.T) { // Initialize the database ctx := context.Background() + db, err := database.New(ctx, dbPath) if err != nil { t.Fatalf("Failed to create database: %v", err) } + defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Logf("Failed to close database: %v", err) } }() @@ -142,12 +150,14 @@ func TestBackupWithInMemoryFS(t *testing.T) { if !expectedFiles[file.Path.String()] { t.Errorf("Unexpected file in database: %s", file.Path) } + delete(expectedFiles, file.Path.String()) // Verify file metadata fsFile := testFS[file.Path.String()] if fsFile == nil { t.Errorf("File %s not found in test filesystem", file.Path) + continue } @@ -187,6 +197,7 @@ func TestBackupWithInMemoryFS(t *testing.T) { if err != nil { t.Fatalf("Failed to get blob hashes: %v", err) } + if len(blobHashes) == 0 { t.Error("Expected at least one blob to be created") } @@ -197,6 +208,7 @@ func TestBackupWithInMemoryFS(t *testing.T) { if err != nil { t.Errorf("Failed to check blob %s: %v", blobHash, err) } + if !exists { t.Errorf("Blob %s not found in S3", blobHash) } @@ -229,12 +241,15 @@ func TestBackupDeduplication(t *testing.T) { // Initialize the database ctx := context.Background() + db, err := database.New(ctx, dbPath) if err != nil { t.Fatalf("Failed to create database: %v", err) } + defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Logf("Failed to close database: %v", err) } }() @@ -348,6 +363,7 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str UID: 1000, // Default UID for test GID: 1000, // Default GID for test } + err = b.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { return b.repos.Files.Create(ctx, tx, file) }) @@ -364,7 +380,8 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str return err } defer func() { - if err := f.Close(); err != nil { + err := f.Close() + if err != nil { // Log but don't fail since we're already in an error path potentially fmt.Fprintf(os.Stderr, "Failed to close file: %v\n", err) } @@ -376,9 +393,10 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str for { n, err := f.Read(buffer) - if err != nil && err != io.EOF { + if err != nil && !errors.Is(err, io.EOF) { return err } + if n == 0 { break } @@ -395,11 +413,13 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str ChunkHash: types.ChunkHash(chunkHash), Size: int64(n), } + return b.repos.Chunks.Create(ctx, tx, chunk) }) if err != nil { return err } + processedChunks[chunkHash] = true } @@ -410,6 +430,7 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str Idx: chunkIndex, ChunkHash: types.ChunkHash(chunkHash), } + return b.repos.FileChunks.Create(ctx, tx, fileChunk) }) if err != nil { @@ -424,6 +445,7 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str FileOffset: int64(chunkIndex * defaultChunkSize), Length: int64(n), } + return b.repos.ChunkFiles.Create(ctx, tx, chunkFile) }) if err != nil { @@ -435,7 +457,6 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str return nil }) - if err != nil { return "", err } @@ -464,12 +485,14 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str // Create blob entry in a short transaction blobID := types.NewBlobID() + err = b.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { blob := &database.Blob{ ID: blobID, Hash: types.BlobHash(blobHash), CreatedTS: time.Now(), } + return b.repos.Blobs.Create(ctx, tx, blob) }) if err != nil { @@ -487,6 +510,7 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str Offset: 0, Length: chunk.Size, } + return b.repos.BlobChunks.Create(ctx, tx, blobChunk) }) if err != nil { @@ -506,7 +530,6 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str err = b.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { return b.repos.Snapshots.UpdateCounts(ctx, tx, snapshotID, fileCount, chunkCount, blobCount, totalSize, blobSize) }) - if err != nil { return "", err } @@ -517,16 +540,18 @@ func (b *BackupEngine) Backup(ctx context.Context, fsys fs.FS, root string) (str func calculateHash(data []byte) string { h := sha256.New() h.Write(data) - return fmt.Sprintf("%x", h.Sum(nil)) + + return hex.EncodeToString(h.Sum(nil)) } func generateLargeFileContent(size int) []byte { data := make([]byte, size) // Fill with pattern that changes every chunk to avoid deduplication - for i := 0; i < size; i++ { + for i := range size { chunkNum := i / defaultChunkSize data[i] = byte((i + chunkNum) % 256) } + return data } diff --git a/internal/snapshot/exclude_test.go b/internal/snapshot/exclude_test.go index ba2efef..f908250 100644 --- a/internal/snapshot/exclude_test.go +++ b/internal/snapshot/exclude_test.go @@ -63,6 +63,7 @@ func setupExcludeTestFS(t *testing.T) afero.Fs { } testTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + for path, content := range files { dir := filepath.Dir(path) err := fs.MkdirAll(dir, 0755) @@ -107,6 +108,7 @@ func createTestScanner(t *testing.T, fs afero.Fs, excludePatterns []string) (*sn func createSnapshotRecord(t *testing.T, ctx context.Context, repos *database.Repositories, snapshotID string) { t.Helper() + err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { snap := &database.Snapshot{ ID: types.SnapshotID(snapshotID), @@ -121,6 +123,7 @@ func createSnapshotRecord(t *testing.T, ctx context.Context, repos *database.Rep BlobSize: 0, CompressionRatio: 1.0, } + return repos.Snapshots.Create(ctx, tx, snap) }) require.NoError(t, err) @@ -128,8 +131,10 @@ func createSnapshotRecord(t *testing.T, ctx context.Context, repos *database.Rep func TestExcludePatterns_ExcludeGitDirectory(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{".git"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -148,8 +153,10 @@ func TestExcludePatterns_ExcludeGitDirectory(t *testing.T) { func TestExcludePatterns_ExcludeByExtension(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"*.log"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -165,8 +172,10 @@ func TestExcludePatterns_ExcludeByExtension(t *testing.T) { func TestExcludePatterns_ExcludeNodeModules(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"node_modules"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -182,8 +191,10 @@ func TestExcludePatterns_ExcludeNodeModules(t *testing.T) { func TestExcludePatterns_MultiplePatterns(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{".git", "node_modules", "*.log", ".DS_Store", "thumbs.db", "cache", "build"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -199,8 +210,10 @@ func TestExcludePatterns_MultiplePatterns(t *testing.T) { func TestExcludePatterns_NoExclusions(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -215,8 +228,10 @@ func TestExcludePatterns_NoExclusions(t *testing.T) { func TestExcludePatterns_ExcludeHiddenFiles(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{".*"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -232,8 +247,10 @@ func TestExcludePatterns_ExcludeHiddenFiles(t *testing.T) { func TestExcludePatterns_DoubleStarGlob(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"**/*.pack"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -249,8 +266,10 @@ func TestExcludePatterns_DoubleStarGlob(t *testing.T) { func TestExcludePatterns_ExactFileName(t *testing.T) { fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"thumbs.db", ".DS_Store"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -267,8 +286,10 @@ func TestExcludePatterns_ExactFileName(t *testing.T) { func TestExcludePatterns_CaseSensitive(t *testing.T) { // Pattern matching should be case-sensitive fs := setupExcludeTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"THUMBS.DB"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -287,6 +308,7 @@ func TestExcludePatterns_DirectoryWithTrailingSlash(t *testing.T) { // Some users might add trailing slashes to directory patterns scanner, repos, cleanup := createTestScanner(t, fs, []string{"cache/", "build/"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -305,6 +327,7 @@ func TestExcludePatterns_PatternInSubdirectory(t *testing.T) { // Exclude .hidden file specifically in src directory scanner, repos, cleanup := createTestScanner(t, fs, []string{"src/.hidden"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -343,6 +366,7 @@ func setupAnchoredTestFS(t *testing.T) afero.Fs { } testTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + for path, content := range files { dir := filepath.Dir(path) err := fs.MkdirAll(dir, 0755) @@ -359,8 +383,10 @@ func setupAnchoredTestFS(t *testing.T) afero.Fs { func TestExcludePatterns_AnchoredPattern(t *testing.T) { // Pattern starting with / should only match from root of source dir fs := setupAnchoredTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"/projectname"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -378,8 +404,10 @@ func TestExcludePatterns_AnchoredPattern(t *testing.T) { func TestExcludePatterns_UnanchoredPattern(t *testing.T) { // Pattern without leading / should match anywhere in path fs := setupAnchoredTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"projectname"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -398,8 +426,10 @@ func TestExcludePatterns_UnanchoredPattern(t *testing.T) { func TestExcludePatterns_AnchoredPatternWithGlob(t *testing.T) { // Anchored pattern with glob fs := setupAnchoredTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"/src/*.go"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -416,8 +446,10 @@ func TestExcludePatterns_AnchoredPatternWithGlob(t *testing.T) { func TestExcludePatterns_AnchoredPatternFile(t *testing.T) { // Anchored pattern for exact file at root fs := setupAnchoredTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"/file.txt"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() @@ -435,8 +467,10 @@ func TestExcludePatterns_AnchoredPatternFile(t *testing.T) { func TestExcludePatterns_UnanchoredPatternFile(t *testing.T) { // Unanchored pattern for file should match anywhere fs := setupAnchoredTestFS(t) + scanner, repos, cleanup := createTestScanner(t, fs, []string{"file.txt"}) defer cleanup() + require.NotNil(t, scanner) ctx := context.Background() diff --git a/internal/snapshot/file_change_test.go b/internal/snapshot/file_change_test.go index 033934b..a918010 100644 --- a/internal/snapshot/file_change_test.go +++ b/internal/snapshot/file_change_test.go @@ -30,9 +30,11 @@ func TestFileContentChange(t *testing.T) { // Create test database db, err := database.NewTestDB() + require.NoError(t, err) defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -59,6 +61,7 @@ func TestFileContentChange(t *testing.T) { VaultikVersion: "test", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) @@ -81,6 +84,7 @@ func TestFileContentChange(t *testing.T) { // Modify the file time.Sleep(10 * time.Millisecond) // Ensure mtime changes + err = afero.WriteFile(fs, "/test.txt", []byte("Modified content with different data"), 0644) require.NoError(t, err) @@ -93,6 +97,7 @@ func TestFileContentChange(t *testing.T) { VaultikVersion: "test", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) @@ -130,6 +135,7 @@ func TestFileContentChange(t *testing.T) { // Verify that chunk_files for old chunk no longer references this file oldChunkFiles, err := repos.ChunkFiles.GetByChunkHash(ctx, oldChunkHash) require.NoError(t, err) + for _, cf := range oldChunkFiles { file, err := repos.Files.GetByID(ctx, cf.FileID) require.NoError(t, err) @@ -159,9 +165,11 @@ func TestMultipleFileChanges(t *testing.T) { // Create test database db, err := database.NewTestDB() + require.NoError(t, err) defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -188,6 +196,7 @@ func TestMultipleFileChanges(t *testing.T) { VaultikVersion: "test", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) @@ -200,6 +209,7 @@ func TestMultipleFileChanges(t *testing.T) { // Modify two files time.Sleep(10 * time.Millisecond) // Ensure mtime changes + err = afero.WriteFile(fs, "/file1.txt", []byte("Modified content 1"), 0644) require.NoError(t, err) err = afero.WriteFile(fs, "/file3.txt", []byte("Modified content 3"), 0644) @@ -214,6 +224,7 @@ func TestMultipleFileChanges(t *testing.T) { VaultikVersion: "test", StartedAt: time.Now(), } + return repos.Snapshots.Create(ctx, tx, snapshot) }) require.NoError(t, err) diff --git a/internal/snapshot/manifest.go b/internal/snapshot/manifest.go index 3b152cc..f6dc3ae 100644 --- a/internal/snapshot/manifest.go +++ b/internal/snapshot/manifest.go @@ -52,6 +52,7 @@ func EncodeManifest(manifest *Manifest, compressionLevel int) ([]byte, error) { // Compress using zstd var compressedBuf bytes.Buffer + writer, err := zstd.NewWriter(&compressedBuf, zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(compressionLevel))) if err != nil { return nil, fmt.Errorf("creating zstd writer: %w", err) @@ -59,6 +60,7 @@ func EncodeManifest(manifest *Manifest, compressionLevel int) ([]byte, error) { if _, err := writer.Write(jsonData); err != nil { _ = writer.Close() + return nil, fmt.Errorf("writing compressed data: %w", err) } diff --git a/internal/snapshot/permission_error_test.go b/internal/snapshot/permission_error_test.go index bbde682..d339d3d 100644 --- a/internal/snapshot/permission_error_test.go +++ b/internal/snapshot/permission_error_test.go @@ -12,7 +12,9 @@ import ( func TestWrapPermissionError(t *testing.T) { // Non-permission errors pass through unchanged. plain := errors.New("disk on fire") - if got := wrapPermissionError("/some/path", plain); got != plain { + + got := wrapPermissionError("/some/path", plain) + if !errors.Is(got, plain) { t.Errorf("non-permission error should pass through, got %v", got) } @@ -23,6 +25,7 @@ func TestWrapPermissionError(t *testing.T) { if !errors.Is(wrapped, os.ErrPermission) { t.Error("wrapped error should still match os.ErrPermission") } + if !strings.Contains(wrapped.Error(), "/Users/u/Library/Calendars") { t.Error("wrapped error should name the offending path") } @@ -31,6 +34,7 @@ func TestWrapPermissionError(t *testing.T) { if !strings.Contains(wrapped.Error(), "Full Disk Access") { t.Errorf("macOS permission error should mention Full Disk Access:\n%s", wrapped.Error()) } + if !strings.Contains(wrapped.Error(), "System Settings") { t.Errorf("macOS permission error should point at System Settings:\n%s", wrapped.Error()) } diff --git a/internal/snapshot/progress.go b/internal/snapshot/progress.go index 568efd2..8795e92 100644 --- a/internal/snapshot/progress.go +++ b/internal/snapshot/progress.go @@ -153,6 +153,7 @@ func (pr *ProgressReporter) printSummaryStatus() { if uploadInfo, ok := pr.stats.CurrentUpload.Load().(*UploadInfo); ok && uploadInfo != nil { // Show upload progress instead pr.printUploadProgress(uploadInfo) + return } @@ -175,16 +176,18 @@ func (pr *ProgressReporter) printSummaryStatus() { // Calculate ETA if we have total size and are processing etaStr := "" + if totalSize > 0 && bytesProcessed > 0 { processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time) if ok && !processStart.IsZero() { processElapsed := time.Since(processStart) + rate := float64(bytesProcessed) / processElapsed.Seconds() if rate > 0 { remainingBytes := totalSize - bytesProcessed remainingSeconds := float64(remainingBytes) / rate eta := time.Duration(remainingSeconds * float64(time.Second)) - etaStr = fmt.Sprintf(" | ETA: %s", formatDuration(eta)) + etaStr = " | ETA: " + formatDuration(eta) } } } @@ -206,7 +209,7 @@ func (pr *ProgressReporter) printSummaryStatus() { ) if currentFile != "" { - status += fmt.Sprintf(" | Current: %s", truncatePath(currentFile, 40)) + status += " | Current: " + truncatePath(currentFile, 40) } log.Info(status) @@ -242,6 +245,7 @@ func (pr *ProgressReporter) printDetailedStatus() { processStart, ok := pr.stats.ProcessStartTime.Load().(time.Time) if ok && !processStart.IsZero() { processElapsed := time.Since(processStart) + processRate := float64(bytesProcessed) / processElapsed.Seconds() if processRate > 0 { remainingBytes := totalSize - bytesProcessed @@ -276,9 +280,11 @@ func (pr *ProgressReporter) printDetailedStatus() { log.Info("Total uploaded to remote", "uploaded", humanize.Bytes(uint64(bytesUploaded)), "compression_ratio", formatRatio(bytesUploaded, bytesScanned)) + if currentFile != "" { log.Info("Current file", "path", currentFile) } + log.Notice("=============================") } @@ -288,12 +294,15 @@ func formatDuration(d time.Duration) string { if d < 0 { return "unknown" } + if d < time.Minute { return fmt.Sprintf("%ds", int(d.Seconds())) } + if d < time.Hour { return fmt.Sprintf("%dm%ds", int(d.Minutes()), int(d.Seconds())%60) } + return fmt.Sprintf("%dh%dm", int(d.Hours()), int(d.Minutes())%60) } @@ -301,6 +310,7 @@ func formatPercent(numerator, denominator int64) string { if denominator == 0 { return "0.0%" } + return fmt.Sprintf("%.1f%%", float64(numerator)/float64(denominator)*100) } @@ -308,7 +318,9 @@ func formatRatio(compressed, uncompressed int64) string { if uncompressed == 0 { return "1.00" } + ratio := float64(compressed) / float64(uncompressed) + return fmt.Sprintf("%.2f", ratio) } @@ -353,6 +365,7 @@ func (pr *ProgressReporter) ReportUploadComplete(blobHash string, size int64, du if duration < time.Millisecond { duration = time.Millisecond } + bytesPerSec := float64(size) / duration.Seconds() bitsPerSec := bytesPerSec * 8 @@ -398,6 +411,7 @@ func (pr *ProgressReporter) ReportUploadProgress(blobHash string, bytesUploaded, // Calculate ETA based on current speed etaStr := "unknown" + if instantSpeed > 0 && bytesUploaded < totalSize { remainingBytes := totalSize - bytesUploaded remainingSeconds := float64(remainingBytes) / instantSpeed diff --git a/internal/snapshot/remotekey.go b/internal/snapshot/remotekey.go index 031b9ce..7338f07 100644 --- a/internal/snapshot/remotekey.go +++ b/internal/snapshot/remotekey.go @@ -36,5 +36,6 @@ const remoteKeyPrefix = "vaultik|" func RemoteSnapshotKey(snapshotID string) string { first := sha256.Sum256([]byte(remoteKeyPrefix + snapshotID)) second := sha256.Sum256(first[:]) + return hex.EncodeToString(second[:]) } diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index c6b2442..2501832 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -119,6 +119,7 @@ func NewScanner(cfg ScannerConfig) *Scanner { // Create encryptor (required for blob packing) if len(cfg.AgeRecipients) == 0 { log.Error("No age recipients configured - encryption is required") + return nil } @@ -130,9 +131,11 @@ func NewScanner(cfg ScannerConfig) *Scanner { Repositories: cfg.Repositories, Fs: cfg.FS, } + packer, err := blob.NewPacker(packerCfg) if err != nil { log.Error("Failed to create packer", "error", err) + return nil } @@ -199,11 +202,14 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc // Phase 1: Scan directory, collect files to process, and track existing files // (builds existingFiles map during walk to avoid double traversal) log.Info("Phase 1/3: Scanning directory structure") + existingFiles := make(map[string]struct{}) + scanResult, err := s.scanPhase(ctx, path, result, existingFiles, knownFiles) if err != nil { return nil, fmt.Errorf("scan phase failed: %w", err) } + filesToProcess := scanResult.FilesToProcess // Phase 1b: Detect deleted files by comparing DB against scanned files @@ -214,7 +220,9 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc // Phase 1c: Associate unchanged files with this snapshot (no new records needed) if len(scanResult.UnchangedFileIDs) > 0 { s.ui.Begin("Associating %s unchanged files with the snapshot.", s.ui.Count(len(scanResult.UnchangedFileIDs))) - if err := s.batchAddFilesToSnapshot(ctx, scanResult.UnchangedFileIDs); err != nil { + + err := s.batchAddFilesToSnapshot(ctx, scanResult.UnchangedFileIDs) + if err != nil { return nil, fmt.Errorf("associating unchanged files: %w", err) } } @@ -226,7 +234,9 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc if len(filesToProcess) > 0 { s.ui.Begin("Backing up %s snapshot source files (chunking, compressing, encrypting, uploading).", s.ui.Count(len(filesToProcess))) log.Info("Phase 2/3: Creating snapshot (chunking, compressing, encrypting, and uploading blobs)") - if err := s.processPhase(ctx, filesToProcess, result); err != nil { + + err := s.processPhase(ctx, filesToProcess, result) + if err != nil { return nil, fmt.Errorf("process phase failed: %w", err) } } else { @@ -244,16 +254,20 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc // This avoids per-file and per-chunk database queries during the scan and process phases func (s *Scanner) loadDatabaseState(ctx context.Context, path string) (map[string]*database.File, error) { s.ui.Begin("Loading known files from local index database.") + knownFiles, err := s.loadKnownFiles(ctx, path) if err != nil { return nil, fmt.Errorf("loading known files: %w", err) } + s.ui.Complete("Loaded %s known files from local index database.", s.ui.Count(len(knownFiles))) s.ui.Begin("Loading known chunks from local index database.") + if err := s.loadKnownChunks(ctx); err != nil { return nil, fmt.Errorf("loading known chunks: %w", err) } + s.ui.Complete("Loaded %s known chunks from local index database.", s.ui.Count(len(s.knownChunks))) return knownFiles, nil @@ -288,6 +302,7 @@ func (s *Scanner) summarizeScanPhase(result *ScanResult, filesToProcess []*FileT s.ui.Count(result.FilesDeleted), s.ui.Size(result.BytesDeleted)) } + s.ui.Complete("%s.", msg) } @@ -337,6 +352,7 @@ func (s *Scanner) loadKnownChunks(ctx context.Context) error { } s.knownChunksMu.Lock() + s.knownChunks = make(map[string]struct{}, len(chunks)) for _, c := range chunks { s.knownChunks[c.ChunkHash.String()] = struct{}{} @@ -351,6 +367,7 @@ func (s *Scanner) chunkExists(hash string) bool { s.knownChunksMu.RLock() _, exists := s.knownChunks[hash] s.knownChunksMu.RUnlock() + return exists } @@ -371,7 +388,9 @@ func (s *Scanner) addPendingChunkHash(hash string) { // removePendingChunkHashes removes committed chunk hashes from the pending set func (s *Scanner) removePendingChunkHashes(hashes []string) { log.Debug("removePendingChunkHashes: starting", "count", len(hashes)) + start := time.Now() + s.pendingChunkHashesMu.Lock() for _, hash := range hashes { delete(s.pendingChunkHashes, hash) @@ -385,6 +404,7 @@ func (s *Scanner) isChunkPending(hash string) bool { s.pendingChunkHashesMu.Lock() _, pending := s.pendingChunkHashes[hash] s.pendingChunkHashesMu.Unlock() + return pending } @@ -411,37 +431,45 @@ func (s *Scanner) flushPendingFiles(ctx context.Context) error { return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error { for _, data := range files { // Create or update the file record - if err := s.repos.Files.Create(txCtx, tx, data.file); err != nil { + err := s.repos.Files.Create(txCtx, tx, data.file) + if err != nil { return fmt.Errorf("creating file record: %w", err) } // Delete any existing file_chunks and chunk_files for this file - if err := s.repos.FileChunks.DeleteByFileID(txCtx, tx, data.file.ID); err != nil { + err = s.repos.FileChunks.DeleteByFileID(txCtx, tx, data.file.ID) + if err != nil { return fmt.Errorf("deleting old file chunks: %w", err) } - if err := s.repos.ChunkFiles.DeleteByFileID(txCtx, tx, data.file.ID); err != nil { + + err = s.repos.ChunkFiles.DeleteByFileID(txCtx, tx, data.file.ID) + if err != nil { return fmt.Errorf("deleting old chunk files: %w", err) } // Create file-chunk mappings for i := range data.fileChunks { - if err := s.repos.FileChunks.Create(txCtx, tx, &data.fileChunks[i]); err != nil { + err := s.repos.FileChunks.Create(txCtx, tx, &data.fileChunks[i]) + if err != nil { return fmt.Errorf("creating file chunk: %w", err) } } // Create chunk-file mappings for i := range data.chunkFiles { - if err := s.repos.ChunkFiles.Create(txCtx, tx, &data.chunkFiles[i]); err != nil { + err := s.repos.ChunkFiles.Create(txCtx, tx, &data.chunkFiles[i]) + if err != nil { return fmt.Errorf("creating chunk file: %w", err) } } // Add file to snapshot - if err := s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, data.file.ID); err != nil { + err = s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, data.file.ID) + if err != nil { return fmt.Errorf("adding file to snapshot: %w", err) } } + return nil }) } @@ -455,6 +483,7 @@ func (s *Scanner) flushAllPending(ctx context.Context) error { // Files with pending chunks are kept in the queue for later flushing func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error { flushStart := time.Now() + log.Debug("flushCompletedPendingFiles: starting") // Partition pending files into those ready to flush and those still waiting @@ -462,6 +491,7 @@ func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error { if len(canFlush) == 0 { log.Debug("flushCompletedPendingFiles: nothing to flush") + return nil } @@ -474,10 +504,13 @@ func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error { // Execute the batch flush in a single transaction log.Debug("flushCompletedPendingFiles: starting transaction") + txStart := time.Now() err := s.executeBatchFileFlush(ctx, allFiles, allFileIDs, allFileChunks, allChunkFiles) + log.Debug("flushCompletedPendingFiles: transaction done", "duration", time.Since(txStart)) log.Debug("flushCompletedPendingFiles: total duration", "duration", time.Since(flushStart)) + return err } @@ -492,21 +525,27 @@ func (s *Scanner) partitionPendingByChunkStatus() (canFlush []pendingFileData, s var stillPending []pendingFileData log.Debug("flushCompletedPendingFiles: checking which files can flush") + checkStart := time.Now() + for _, data := range s.pendingFiles { allChunksCommitted := true + for _, fc := range data.fileChunks { if s.isChunkPending(fc.ChunkHash.String()) { allChunksCommitted = false + break } } + if allChunksCommitted { canFlush = append(canFlush, data) } else { stillPending = append(stillPending, data) } } + log.Debug("flushCompletedPendingFiles: check done", "duration", time.Since(checkStart), "can_flush", len(canFlush), "still_pending", len(stillPending)) s.pendingFiles = stillPending @@ -520,12 +559,15 @@ func (s *Scanner) partitionPendingByChunkStatus() (canFlush []pendingFileData, s // mappings from the given pending file data for efficient batch database operations func (s *Scanner) collectBatchFlushData(canFlush []pendingFileData) ([]*database.File, []types.FileID, []database.FileChunk, []database.ChunkFile) { log.Debug("flushCompletedPendingFiles: collecting data for batch ops") + collectStart := time.Now() - var allFileChunks []database.FileChunk - var allChunkFiles []database.ChunkFile - var allFileIDs []types.FileID - var allFiles []*database.File + var ( + allFileChunks []database.FileChunk + allChunkFiles []database.ChunkFile + allFileIDs []types.FileID + allFiles []*database.File + ) for _, data := range canFlush { allFileChunks = append(allFileChunks, data.fileChunks...) @@ -551,52 +593,77 @@ func (s *Scanner) executeBatchFileFlush(ctx context.Context, allFiles []*databas // Batch delete old file_chunks and chunk_files log.Debug("flushCompletedPendingFiles: deleting old file_chunks") + opStart := time.Now() - if err := s.repos.FileChunks.DeleteByFileIDs(txCtx, tx, allFileIDs); err != nil { + + err := s.repos.FileChunks.DeleteByFileIDs(txCtx, tx, allFileIDs) + if err != nil { return fmt.Errorf("batch deleting old file chunks: %w", err) } + log.Debug("flushCompletedPendingFiles: deleted file_chunks", "duration", time.Since(opStart)) log.Debug("flushCompletedPendingFiles: deleting old chunk_files") + opStart = time.Now() - if err := s.repos.ChunkFiles.DeleteByFileIDs(txCtx, tx, allFileIDs); err != nil { + + err = s.repos.ChunkFiles.DeleteByFileIDs(txCtx, tx, allFileIDs) + if err != nil { return fmt.Errorf("batch deleting old chunk files: %w", err) } + log.Debug("flushCompletedPendingFiles: deleted chunk_files", "duration", time.Since(opStart)) // Batch create/update file records log.Debug("flushCompletedPendingFiles: creating files") + opStart = time.Now() - if err := s.repos.Files.CreateBatch(txCtx, tx, allFiles); err != nil { + + err = s.repos.Files.CreateBatch(txCtx, tx, allFiles) + if err != nil { return fmt.Errorf("batch creating file records: %w", err) } + log.Debug("flushCompletedPendingFiles: created files", "duration", time.Since(opStart)) // Batch insert file_chunks log.Debug("flushCompletedPendingFiles: inserting file_chunks") + opStart = time.Now() - if err := s.repos.FileChunks.CreateBatch(txCtx, tx, allFileChunks); err != nil { + + err = s.repos.FileChunks.CreateBatch(txCtx, tx, allFileChunks) + if err != nil { return fmt.Errorf("batch creating file chunks: %w", err) } + log.Debug("flushCompletedPendingFiles: inserted file_chunks", "duration", time.Since(opStart)) // Batch insert chunk_files log.Debug("flushCompletedPendingFiles: inserting chunk_files") + opStart = time.Now() - if err := s.repos.ChunkFiles.CreateBatch(txCtx, tx, allChunkFiles); err != nil { + + err = s.repos.ChunkFiles.CreateBatch(txCtx, tx, allChunkFiles) + if err != nil { return fmt.Errorf("batch creating chunk files: %w", err) } + log.Debug("flushCompletedPendingFiles: inserted chunk_files", "duration", time.Since(opStart)) // Batch add files to snapshot log.Debug("flushCompletedPendingFiles: adding files to snapshot") + opStart = time.Now() - if err := s.repos.Snapshots.AddFilesByIDBatch(txCtx, tx, s.snapshotID, allFileIDs); err != nil { + + err = s.repos.Snapshots.AddFilesByIDBatch(txCtx, tx, s.snapshotID, allFileIDs) + if err != nil { return fmt.Errorf("batch adding files to snapshot: %w", err) } + log.Debug("flushCompletedPendingFiles: added files to snapshot", "duration", time.Since(opStart)) log.Debug("flushCompletedPendingFiles: transaction complete") + return nil }) } @@ -616,24 +683,31 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult estimatedTotal := int64(len(knownFiles)) var filesToProcess []*FileToProcess + var unchangedFileIDs []types.FileID // Just IDs - no new records needed + var mu sync.Mutex // Set up periodic status output startTime := time.Now() lastStatusTime := time.Now() statusInterval := 15 * time.Second + var filesScanned int64 log.Debug("Starting directory walk", "path", path) + err := afero.Walk(s.fs, path, func(filePath string, info os.FileInfo, err error) error { if err != nil { if s.skipErrors { log.Error("Failed to access file (skipping due to --skip-errors)", "path", filePath, "error", err) s.ui.Error("Failed to access %s: %v. Skipping (--skip-errors).", s.ui.Path(filePath), err) + return nil // Continue scanning } + log.Debug("Error accessing filesystem entry", "path", filePath, "error", err) + return wrapPermissionError(filePath, err) } @@ -649,6 +723,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult if info.IsDir() { return filepath.SkipDir } + return nil } @@ -657,7 +732,9 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult file := s.buildSymlinkEntry(filePath, info) if file != nil { existingFiles[filePath] = struct{}{} + mu.Lock() + filesToProcess = append(filesToProcess, &FileToProcess{ Path: filePath, FileInfo: info, @@ -667,6 +744,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult mu.Unlock() s.updateScanEntryStats(result, true, info) } + return nil } @@ -674,7 +752,9 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult if info.IsDir() { file := s.buildDirectoryEntry(filePath, info) existingFiles[filePath] = struct{}{} + mu.Lock() + filesToProcess = append(filesToProcess, &FileToProcess{ Path: filePath, FileInfo: info, @@ -682,6 +762,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult }) filesScanned++ mu.Unlock() + return nil } @@ -708,6 +789,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult // Unchanged file with existing ID - just need snapshot association unchangedFileIDs = append(unchangedFileIDs, file.ID) } + filesScanned++ changedCount := len(filesToProcess) mu.Unlock() @@ -718,12 +800,12 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult // Output periodic status if time.Since(lastStatusTime) >= statusInterval { s.printScanProgressLine(filesScanned, changedCount, estimatedTotal, startTime) + lastStatusTime = time.Now() } return nil }) - if err != nil { return nil, err } @@ -745,11 +827,13 @@ func (s *Scanner) updateScanEntryStats(result *ScanResult, needsProcessing bool, } else { result.FilesSkipped++ result.BytesSkipped += info.Size() + if s.progress != nil { s.progress.GetStats().FilesSkipped.Add(1) s.progress.GetStats().BytesSkipped.Add(info.Size()) } } + result.FilesScanned++ if s.progress != nil { s.progress.GetStats().FilesScanned.Add(1) @@ -768,14 +852,14 @@ func (s *Scanner) printScanProgressLine(filesScanned int64, changedCount int, es if pct > 100 { pct = 100 // Cap at 100% for display } - remaining := estimatedTotal - filesScanned - if remaining < 0 { - remaining = 0 - } + + remaining := max(estimatedTotal-filesScanned, 0) + var eta time.Duration if rate > 0 && remaining > 0 { eta = time.Duration(float64(remaining)/rate) * time.Second } + if eta > 0 { s.ui.Progress("Snapshot source files enumeration: %s files (~%s), %s changed or new, %.0f files/sec, enumeration elapsed: %s, enumeration ETA: %s (est remain %s).", s.ui.Count(int(filesScanned)), @@ -808,6 +892,7 @@ func (s *Scanner) buildSymlinkEntry(path string, info os.FileInfo) *database.Fil target, err := os.Readlink(path) if err != nil { log.Debug("Cannot read symlink target", "path", path, "error", err) + return nil } @@ -860,9 +945,11 @@ func (s *Scanner) buildDirectoryEntry(path string, info os.FileInfo) *database.F // and associates it with the current snapshot. No chunking is performed. func (s *Scanner) recordNonRegularFile(ctx context.Context, ftp *FileToProcess) error { return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error { - if err := s.repos.Files.Create(txCtx, tx, ftp.File); err != nil { + err := s.repos.Files.Create(txCtx, tx, ftp.File) + if err != nil { return fmt.Errorf("creating non-regular file record: %w", err) } + return s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, ftp.File.ID) }) } @@ -941,18 +1028,18 @@ func (s *Scanner) batchAddFilesToSnapshot(ctx context.Context, fileIDs []types.F default: } - end := i + batchSize - if end > len(fileIDs) { - end = len(fileIDs) - } + end := min(i+batchSize, len(fileIDs)) + batch := fileIDs[i:end] err := s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { for _, fileID := range batch { - if err := s.repos.Snapshots.AddFileByID(ctx, tx, s.snapshotID, fileID); err != nil { + err := s.repos.Snapshots.AddFileByID(ctx, tx, s.snapshotID, fileID) + if err != nil { return fmt.Errorf("adding file to snapshot: %w", err) } } + return nil }) if err != nil { @@ -966,6 +1053,7 @@ func (s *Scanner) batchAddFilesToSnapshot(ctx context.Context, fileIDs []types.F pct := float64(end) / float64(len(fileIDs)) * 100 s.ui.Progress("Snapshot unchanged-file association: %s/%s (%s), %.0f files/sec.", s.ui.Count(end), s.ui.Count(len(fileIDs)), s.ui.Percent(pct), rate) + lastStatusTime = time.Now() } } @@ -991,7 +1079,9 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc statusInterval := 15 * time.Second startTime := time.Now() filesProcessed := 0 + var bytesProcessed int64 + totalFiles := len(filesToProcess) // Process each file @@ -1006,6 +1096,7 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc if err != nil { return err } + if skipped { continue } @@ -1021,6 +1112,7 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc // Output periodic status if time.Since(lastStatusTime) >= statusInterval { s.printProcessingProgress(filesProcessed, totalFiles, bytesProcessed, totalBytes, startTime) + lastStatusTime = time.Now() } } @@ -1032,22 +1124,29 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc // processFileWithErrorHandling wraps processFileStreaming with error recovery for // deleted files and skip-errors mode. Returns (skipped, error). func (s *Scanner) processFileWithErrorHandling(ctx context.Context, fileToProcess *FileToProcess, result *ScanResult) (bool, error) { - if err := s.processFileStreaming(ctx, fileToProcess, result); err != nil { + err := s.processFileStreaming(ctx, fileToProcess, result) + if err != nil { // Handle files that were deleted between scan and process phases if errors.Is(err, os.ErrNotExist) { log.Warn("File was deleted during backup, skipping", "path", fileToProcess.Path) + result.FilesSkipped++ + return true, nil } // Skip file read errors if --skip-errors is enabled if s.skipErrors { log.Error("Failed to process file (skipping due to --skip-errors)", "path", fileToProcess.Path, "error", err) s.ui.Error("Failed to process %s: %v. Skipping (--skip-errors).", s.ui.Path(fileToProcess.Path), err) + result.FilesSkipped++ + return true, nil } + return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err) } + return false, nil } @@ -1061,6 +1160,7 @@ func (s *Scanner) printProcessingProgress(filesProcessed, totalFiles int, bytesP // Calculate ETA based on bytes (more accurate than files) remainingBytes := totalBytes - bytesProcessed + var eta time.Duration if byteRate > 0 { eta = time.Duration(float64(remainingBytes)/byteRate) * time.Second @@ -1097,15 +1197,19 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult) // Final packer flush first - this commits remaining chunks to DB // and handleBlobReady will flush files whose chunks are now committed s.packerMu.Lock() - if err := s.packer.Flush(); err != nil { + + err := s.packer.Flush() + if err != nil { s.packerMu.Unlock() + return fmt.Errorf("flushing packer: %w", err) } s.packerMu.Unlock() // Flush any remaining pending files (e.g., files with only pre-existing chunks // that didn't trigger a blob finalize) - if err := s.flushAllPending(ctx); err != nil { + err = s.flushAllPending(ctx) + if err != nil { return fmt.Errorf("flushing remaining pending files: %w", err) } @@ -1119,6 +1223,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult) if err != nil { return fmt.Errorf("parsing blob ID: %w", err) } + err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { return s.repos.Snapshots.AddBlob(ctx, tx, s.snapshotID, blobID, types.BlobHash(b.Hash)) }) @@ -1126,6 +1231,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult) return fmt.Errorf("storing blob metadata: %w", err) } } + result.BlobsCreated += len(blobs) } @@ -1148,14 +1254,17 @@ func (s *Scanner) handleBlobReady(blobWithReader *blob.BlobWithReader) error { } blobPath := fmt.Sprintf("blobs/%s/%s/%s", finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash) + blobExists, err := s.uploadBlobIfNeeded(ctx, blobPath, blobWithReader, startTime) if err != nil { s.cleanupBlobTempFile(blobWithReader) + return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err) } if err := s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime); err != nil { s.cleanupBlobTempFile(blobWithReader) + return err } @@ -1183,6 +1292,7 @@ func (s *Scanner) uploadBlobIfNeeded(ctx context.Context, blobPath string, blobW "hash", finishedBlob.Hash, "size", humanize.Bytes(uint64(finishedBlob.Compressed))) s.ui.Info("Blob %s (%s) already exists at %s. Skipping upload.", s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed), s.ui.Path(destination)) + return true, nil } @@ -1191,8 +1301,10 @@ func (s *Scanner) uploadBlobIfNeeded(ctx context.Context, blobPath string, blobW progressCallback := s.makeUploadProgressCallback(ctx, finishedBlob, startTime) - if err := s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader, finishedBlob.Compressed, progressCallback); err != nil { + err := s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader, finishedBlob.Compressed, progressCallback) + if err != nil { log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err) + return false, fmt.Errorf("uploading blob to storage: %w", err) } @@ -1228,17 +1340,21 @@ func (s *Scanner) makeUploadProgressCallback(ctx context.Context, finishedBlob * lastProgressTime := time.Now() lastProgressBytes := int64(0) lastStdoutTime := time.Now() + const stdoutInterval = 15 * time.Second return func(uploaded int64) error { now := time.Now() + elapsed := now.Sub(lastProgressTime).Seconds() if elapsed > 0.5 { bytesSinceLastUpdate := uploaded - lastProgressBytes + speed := float64(bytesSinceLastUpdate) / elapsed if s.progress != nil { s.progress.ReportUploadProgress(finishedBlob.Hash, uploaded, finishedBlob.Compressed, speed) } + lastProgressTime = now lastProgressBytes = uploaded } @@ -1248,10 +1364,12 @@ func (s *Scanner) makeUploadProgressCallback(ctx context.Context, finishedBlob * totalElapsed := now.Sub(uploadStart) pct := float64(uploaded) / float64(finishedBlob.Compressed) * 100 avgSpeed := float64(uploaded) / totalElapsed.Seconds() + var eta time.Duration if avgSpeed > 0 { eta = time.Duration(float64(finishedBlob.Compressed-uploaded)/avgSpeed) * time.Second } + s.ui.Progress("Blob upload %s: %s / %s (%s) at %s, blob upload elapsed: %s, blob upload ETA: %s (est remain %s).", s.ui.Hex(finishedBlob.Hash), s.ui.Size(uploaded), @@ -1283,11 +1401,13 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin uploadDuration := time.Since(startTime) return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error { - if err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID); err != nil { + err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID) + if err != nil { return fmt.Errorf("updating blob upload timestamp: %w", err) } - if err := s.repos.Snapshots.AddBlob(txCtx, tx, s.snapshotID, finishedBlobID, types.BlobHash(finishedBlob.Hash)); err != nil { + err = s.repos.Snapshots.AddBlob(txCtx, tx, s.snapshotID, finishedBlobID, types.BlobHash(finishedBlob.Hash)) + if err != nil { return fmt.Errorf("adding blob to snapshot: %w", err) } @@ -1299,7 +1419,9 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin Size: finishedBlob.Compressed, DurationMs: uploadDuration.Milliseconds(), } - if err := s.repos.Uploads.Create(txCtx, tx, upload); err != nil { + + err := s.repos.Uploads.Create(txCtx, tx, upload) + if err != nil { return fmt.Errorf("recording upload metrics: %w", err) } } @@ -1312,10 +1434,14 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.BlobWithReader) { if blobWithReader.TempFile != nil { tempName := blobWithReader.TempFile.Name() - if err := blobWithReader.TempFile.Close(); err != nil { + + err := blobWithReader.TempFile.Close() + if err != nil { log.Fatal("Failed to close temp file", "file", tempName, "error", err) } - if err := s.fs.Remove(tempName); err != nil { + + err = s.fs.Remove(tempName) + if err != nil { log.Fatal("Failed to remove temp file", "file", tempName, "error", err) } } @@ -1343,6 +1469,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT defer func() { _ = file.Close() }() var chunks []streamingChunkInfo + chunkIndex := 0 fileHash, err := s.chunker.ChunkReaderStreaming(file, func(chunk chunker.Chunk) error { @@ -1372,16 +1499,17 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT s.updateChunkStats(chunkExists, chunk.Size, result) if !chunkExists { - if err := s.addChunkToPacker(chunk); err != nil { + err := s.addChunkToPacker(chunk) + if err != nil { return err } } chunk.Data = nil chunkIndex++ + return nil }) - if err != nil { return fmt.Errorf("chunking file: %w", err) } @@ -1390,6 +1518,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT "path", fileToProcess.Path, "file_hash", fileHash, "chunks", len(chunks)) s.queueFileForBatchInsert(ctx, fileToProcess, chunks) + return nil } @@ -1397,6 +1526,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *ScanResult) { if chunkExists { result.FilesSkipped++ + result.BytesSkipped += chunkSize if s.progress != nil { s.progress.GetStats().BytesSkipped.Add(chunkSize) @@ -1404,6 +1534,7 @@ func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *Sc } else { result.ChunksCreated++ result.BytesScanned += chunkSize + if s.progress != nil { s.progress.GetStats().ChunksCreated.Add(1) s.progress.GetStats().BytesProcessed.Add(chunkSize) @@ -1415,27 +1546,36 @@ func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *Sc // addChunkToPacker adds a chunk to the blob packer, finalizing the current blob if needed func (s *Scanner) addChunkToPacker(chunk chunker.Chunk) error { s.packerMu.Lock() + err := s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data}) - if err == blob.ErrBlobSizeLimitExceeded { - if err := s.packer.FinalizeBlob(); err != nil { + if errors.Is(err, blob.ErrBlobSizeLimitExceeded) { + err := s.packer.FinalizeBlob() + if err != nil { s.packerMu.Unlock() + return fmt.Errorf("finalizing blob: %w", err) } - if err := s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data}); err != nil { + + err = s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data}) + if err != nil { s.packerMu.Unlock() + return fmt.Errorf("adding chunk after finalize: %w", err) } } else if err != nil { s.packerMu.Unlock() + return fmt.Errorf("adding chunk to packer: %w", err) } s.packerMu.Unlock() + return nil } // queueFileForBatchInsert builds file/chunk associations and queues the file for batch DB insert func (s *Scanner) queueFileForBatchInsert(ctx context.Context, fileToProcess *FileToProcess, chunks []streamingChunkInfo) { fileChunks := make([]database.FileChunk, len(chunks)) + chunkFiles := make([]database.ChunkFile, len(chunks)) for i, ci := range chunks { fileChunks[i] = database.FileChunk{ @@ -1503,6 +1643,7 @@ func wrapPermissionError(path string, err error) error { if !errors.Is(err, os.ErrPermission) { return err } + if runtime.GOOS == "darwin" { return fmt.Errorf("cannot read %s: %w\n\n"+ "macOS is blocking access to this path. Grant Full Disk Access to your\n"+ @@ -1510,12 +1651,14 @@ func wrapPermissionError(path string, err error) error { " System Settings → Privacy & Security → Full Disk Access\n\n"+ "then quit and reopen the terminal and re-run the backup", path, err) } + return fmt.Errorf("cannot read %s: %w (check file permissions, or run with --skip-errors to continue past unreadable files)", path, err) } // compileExcludePatterns compiles the exclude patterns into glob matchers func compileExcludePatterns(patterns []string) []compiledPattern { var compiled []compiledPattern + for _, p := range patterns { if p == "" { continue @@ -1523,6 +1666,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern { // Check if pattern is anchored (starts with /) anchored := strings.HasPrefix(p, "/") + pattern := p if anchored { pattern = p[1:] // Remove leading / @@ -1537,6 +1681,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern { g, err := glob.Compile(pattern, '/') if err != nil { log.Warn("Invalid exclude pattern, skipping", "pattern", p, "error", err) + continue } @@ -1546,6 +1691,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern { original: p, }) } + return compiled } diff --git a/internal/snapshot/scanner_test.go b/internal/snapshot/scanner_test.go index 4ceaf84..4ed2d3b 100644 --- a/internal/snapshot/scanner_test.go +++ b/internal/snapshot/scanner_test.go @@ -33,16 +33,22 @@ func TestScannerSimpleDirectory(t *testing.T) { // Create files with specific times testTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + for path, content := range testFiles { dir := filepath.Dir(path) - if err := fs.MkdirAll(dir, 0755); err != nil { + + err := fs.MkdirAll(dir, 0755) + if err != nil { t.Fatalf("failed to create directory %s: %v", dir, err) } - if err := afero.WriteFile(fs, path, []byte(content), 0644); err != nil { + + err = afero.WriteFile(fs, path, []byte(content), 0644) + if err != nil { t.Fatalf("failed to write file %s: %v", path, err) } // Set times - if err := fs.Chtimes(path, testTime, testTime); err != nil { + err = fs.Chtimes(path, testTime, testTime) + if err != nil { t.Fatalf("failed to set times for %s: %v", path, err) } } @@ -53,7 +59,8 @@ func TestScannerSimpleDirectory(t *testing.T) { t.Fatalf("failed to create test database: %v", err) } defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -73,6 +80,7 @@ func TestScannerSimpleDirectory(t *testing.T) { // Create a snapshot record for testing ctx := context.Background() snapshotID := "test-snapshot-001" + err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { snapshot := &database.Snapshot{ ID: types.SnapshotID(snapshotID), @@ -87,6 +95,7 @@ func TestScannerSimpleDirectory(t *testing.T) { BlobSize: 0, CompressionRatio: 1.0, } + return repos.Snapshots.Create(ctx, tx, snapshot) }) if err != nil { @@ -95,6 +104,7 @@ func TestScannerSimpleDirectory(t *testing.T) { // Scan the directory var result *snapshot.ScanResult + result, err = scanner.Scan(ctx, "/source", snapshotID) if err != nil { t.Fatalf("scan failed: %v", err) @@ -170,7 +180,7 @@ func TestScannerLargeFile(t *testing.T) { // Use random content to ensure good chunk boundaries largeContent := make([]byte, 1024*1024) // 1MB // Fill with pseudo-random data to ensure chunk boundaries - for i := 0; i < len(largeContent); i++ { + for i := range largeContent { // Simple pseudo-random generator for deterministic tests largeContent[i] = byte((i * 7919) ^ (i >> 3)) } @@ -178,6 +188,7 @@ func TestScannerLargeFile(t *testing.T) { if err := fs.MkdirAll("/source", 0755); err != nil { t.Fatal(err) } + if err := afero.WriteFile(fs, "/source/large.bin", largeContent, 0644); err != nil { t.Fatal(err) } @@ -188,7 +199,8 @@ func TestScannerLargeFile(t *testing.T) { t.Fatalf("failed to create test database: %v", err) } defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -208,6 +220,7 @@ func TestScannerLargeFile(t *testing.T) { // Create a snapshot record for testing ctx := context.Background() snapshotID := "test-snapshot-001" + err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { snapshot := &database.Snapshot{ ID: types.SnapshotID(snapshotID), @@ -222,6 +235,7 @@ func TestScannerLargeFile(t *testing.T) { BlobSize: 0, CompressionRatio: 1.0, } + return repos.Snapshots.Create(ctx, tx, snapshot) }) if err != nil { @@ -230,6 +244,7 @@ func TestScannerLargeFile(t *testing.T) { // Scan the directory var result *snapshot.ScanResult + result, err = scanner.Scan(ctx, "/source", snapshotID) if err != nil { t.Fatalf("scan failed: %v", err) diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go index f3f26d6..545d6e7 100644 --- a/internal/snapshot/snapshot.go +++ b/internal/snapshot/snapshot.go @@ -37,6 +37,8 @@ import ( "bytes" "context" "database/sql" + "encoding/hex" + "errors" "fmt" "io" "os/exec" @@ -97,12 +99,13 @@ func (sm *SnapshotManager) CreateSnapshot(ctx context.Context, hostname, version func (sm *SnapshotManager) CreateSnapshotWithName(ctx context.Context, hostname, name, version, gitRevision string) (string, error) { // Use short hostname (strip domain if present) shortHostname := hostname - if idx := strings.Index(hostname, "."); idx != -1 { - shortHostname = hostname[:idx] + if before, _, ok := strings.Cut(hostname, "."); ok { + shortHostname = before } // Build snapshot ID with optional name timestamp := time.Now().UTC().Format("2006-01-02T15:04:05Z") + var snapshotID string if name != "" { snapshotID = fmt.Sprintf("%s_%s_%s", shortHostname, name, timestamp) @@ -128,12 +131,12 @@ func (sm *SnapshotManager) CreateSnapshotWithName(ctx context.Context, hostname, err := sm.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { return sm.repos.Snapshots.Create(ctx, tx, snapshot) }) - if err != nil { return "", fmt.Errorf("creating snapshot: %w", err) } log.Info("Created snapshot", "snapshot_id", snapshotID) + return snapshotID, nil } @@ -148,7 +151,6 @@ func (sm *SnapshotManager) UpdateSnapshotStats(ctx context.Context, snapshotID s stats.BytesUploaded, ) }) - if err != nil { return fmt.Errorf("updating snapshot stats: %w", err) } @@ -161,13 +163,14 @@ func (sm *SnapshotManager) UpdateSnapshotStats(ctx context.Context, snapshotID s func (sm *SnapshotManager) UpdateSnapshotStatsExtended(ctx context.Context, snapshotID string, stats ExtendedBackupStats) error { return sm.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { // First update basic stats - if err := sm.repos.Snapshots.UpdateCounts(ctx, tx, snapshotID, + err := sm.repos.Snapshots.UpdateCounts(ctx, tx, snapshotID, int64(stats.FilesScanned), int64(stats.ChunksCreated), int64(stats.BlobsCreated), stats.BytesScanned, stats.BytesUploaded, - ); err != nil { + ) + if err != nil { return err } @@ -190,18 +193,20 @@ func (sm *SnapshotManager) CompleteSnapshot(ctx context.Context, snapshotID stri if err != nil { return err } + if added > 0 { log.Info("Populated snapshot_blobs with dedup-referenced blobs", "snapshot_id", snapshotID, "added", added) } + return sm.repos.Snapshots.MarkComplete(ctx, tx, snapshotID) }) - if err != nil { return fmt.Errorf("marking snapshot complete: %w", err) } log.Info("Completed snapshot", "snapshot_id", snapshotID) + return nil } @@ -229,10 +234,13 @@ func (sm *SnapshotManager) ExportSnapshotMetadata(ctx context.Context, dbPath st if err != nil { return fmt.Errorf("creating temp dir: %w", err) } + log.Debug("Created temporary directory", "path", tempDir) defer func() { log.Debug("Cleaning up temporary directory", "path", tempDir) - if err := sm.fs.RemoveAll(tempDir); err != nil { + + err := sm.fs.RemoveAll(tempDir) + if err != nil { log.Debug("Failed to remove temp dir", "path", tempDir, "error", err) } }() @@ -258,6 +266,7 @@ func (sm *SnapshotManager) ExportSnapshotMetadata(ctx context.Context, dbPath st "snapshot_id", snapshotID, "db_size", len(finalData), "manifest_size", len(blobManifest)) + return nil } @@ -268,17 +277,21 @@ func (sm *SnapshotManager) prepareExportDB(ctx context.Context, dbPath, snapshot // The main database should be closed at this point tempDBPath := filepath.Join(tempDir, "snapshot.db") log.Debug("Copying database to temporary location", "source", dbPath, "destination", tempDBPath) + if err := sm.copyFile(dbPath, tempDBPath); err != nil { return nil, "", fmt.Errorf("copying database: %w", err) } + log.Debug("Database copy complete", "size", sm.getFileSize(tempDBPath)) // Step 2: Clean the temp database to only contain current snapshot data log.Debug("Cleaning temporary database", "snapshot_id", snapshotID) + stats, err := sm.cleanSnapshotDB(ctx, tempDBPath, snapshotID) if err != nil { return nil, "", fmt.Errorf("cleaning snapshot database: %w", err) } + log.Info("Temporary database cleanup complete", "db_path", tempDBPath, "size_after_clean", humanize.Bytes(uint64(sm.getFileSize(tempDBPath))), @@ -294,6 +307,7 @@ func (sm *SnapshotManager) prepareExportDB(ctx context.Context, dbPath, snapshot if err := sm.vacuumDatabase(tempDBPath); err != nil { return nil, "", fmt.Errorf("vacuuming database: %w", err) } + log.Debug("Database vacuumed", "size", humanize.Bytes(uint64(sm.getFileSize(tempDBPath)))) // Step 4: Compress and encrypt the binary database file @@ -301,6 +315,7 @@ func (sm *SnapshotManager) prepareExportDB(ctx context.Context, dbPath, snapshot if err := sm.compressFile(tempDBPath, compressedPath); err != nil { return nil, "", fmt.Errorf("compressing database: %w", err) } + log.Debug("Compression complete", "original_size", humanize.Bytes(uint64(sm.getFileSize(tempDBPath))), "compressed_size", humanize.Bytes(uint64(sm.getFileSize(compressedPath)))) @@ -327,9 +342,12 @@ func (sm *SnapshotManager) uploadSnapshotArtifacts(ctx context.Context, snapshot dbKey := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey) dbUploadStart := time.Now() - if err := sm.storage.Put(ctx, dbKey, bytes.NewReader(dbData)); err != nil { + + err := sm.storage.Put(ctx, dbKey, bytes.NewReader(dbData)) + if err != nil { return fmt.Errorf("uploading snapshot database: %w", err) } + dbUploadDuration := time.Since(dbUploadStart) dbUploadSpeed := float64(len(dbData)) * 8 / dbUploadDuration.Seconds() // bits per second log.Info("Uploaded snapshot database", @@ -341,9 +359,12 @@ func (sm *SnapshotManager) uploadSnapshotArtifacts(ctx context.Context, snapshot // Upload blob manifest (compressed only, not encrypted) manifestKey := fmt.Sprintf("metadata/%s/manifest.json.zst", remoteKey) manifestUploadStart := time.Now() - if err := sm.storage.Put(ctx, manifestKey, bytes.NewReader(manifestData)); err != nil { + + err = sm.storage.Put(ctx, manifestKey, bytes.NewReader(manifestData)) + if err != nil { return fmt.Errorf("uploading blob manifest: %w", err) } + manifestUploadDuration := time.Since(manifestUploadStart) manifestUploadSpeed := float64(len(manifestData)) * 8 / manifestUploadDuration.Seconds() // bits per second log.Info("Uploaded blob manifest", @@ -383,7 +404,8 @@ func (sm *SnapshotManager) cleanSnapshotDB(ctx context.Context, dbPath string, s return nil, fmt.Errorf("opening temp database: %w", err) } defer func() { - if err := db.Close(); err != nil { + err := db.Close() + if err != nil { log.Debug("Failed to close temp database", "error", err) } }() @@ -394,7 +416,8 @@ func (sm *SnapshotManager) cleanSnapshotDB(ctx context.Context, dbPath string, s return nil, fmt.Errorf("beginning transaction: %w", err) } defer func() { - if rbErr := tx.Rollback(); rbErr != nil && rbErr != sql.ErrTxDone { + rbErr := tx.Rollback() + if rbErr != nil && !errors.Is(rbErr, sql.ErrTxDone) { log.Debug("Failed to rollback transaction", "error", rbErr) } }() @@ -430,6 +453,7 @@ func (sm *SnapshotManager) cleanSnapshotDB(ctx context.Context, dbPath string, s // Commit transaction log.Debug("[Temp DB Cleanup] Committing cleanup transaction") + if err := tx.Commit(); err != nil { return nil, fmt.Errorf("committing transaction: %w", err) } @@ -439,23 +463,30 @@ func (sm *SnapshotManager) cleanSnapshotDB(ctx context.Context, dbPath string, s // Count files var fileCount int + err = db.QueryRowWithLog(ctx, "SELECT COUNT(*) FROM files").Scan(&fileCount) if err != nil { return nil, fmt.Errorf("counting files: %w", err) } + stats.FileCount = fileCount // Count chunks var chunkCount int + err = db.QueryRowWithLog(ctx, "SELECT COUNT(*) FROM chunks").Scan(&chunkCount) if err != nil { return nil, fmt.Errorf("counting chunks: %w", err) } + stats.ChunkCount = chunkCount // Count blobs and get sizes - var blobCount int - var compressedSize, uncompressedSize sql.NullInt64 + var ( + blobCount int + compressedSize, uncompressedSize sql.NullInt64 + ) + err = db.QueryRowWithLog(ctx, ` SELECT COUNT(*), COALESCE(SUM(compressed_size), 0), COALESCE(SUM(uncompressed_size), 0) FROM blobs @@ -464,6 +495,7 @@ func (sm *SnapshotManager) cleanSnapshotDB(ctx context.Context, dbPath string, s if err != nil { return nil, fmt.Errorf("counting blobs and sizes: %w", err) } + stats.BlobCount = blobCount stats.CompressedSize = compressedSize.Int64 stats.UncompressedSize = uncompressedSize.Int64 @@ -491,7 +523,8 @@ func (sm *SnapshotManager) compressFile(inputPath, outputPath string) error { return fmt.Errorf("opening input file: %w", err) } defer func() { - if err := input.Close(); err != nil { + err := input.Close() + if err != nil { log.Debug("Failed to close input file", "path", inputPath, "error", err) } }() @@ -501,13 +534,15 @@ func (sm *SnapshotManager) compressFile(inputPath, outputPath string) error { return fmt.Errorf("creating output file: %w", err) } defer func() { - if err := output.Close(); err != nil { + err := output.Close() + if err != nil { log.Debug("Failed to close output file", "path", outputPath, "error", err) } }() // Use blobgen for compression and encryption log.Debug("Compressing and encrypting data") + writer, err := blobgen.NewWriter(output, sm.config.CompressionLevel, sm.config.AgeRecipients) if err != nil { return fmt.Errorf("creating blobgen writer: %w", err) @@ -517,7 +552,8 @@ func (sm *SnapshotManager) compressFile(inputPath, outputPath string) error { writerClosed := false defer func() { if !writerClosed { - if err := writer.Close(); err != nil { + err := writer.Close() + if err != nil { log.Debug("Failed to close writer", "error", err) } } @@ -531,9 +567,10 @@ func (sm *SnapshotManager) compressFile(inputPath, outputPath string) error { if err := writer.Close(); err != nil { return fmt.Errorf("closing writer: %w", err) } + writerClosed = true - log.Debug("Compression complete", "hash", fmt.Sprintf("%x", writer.Sum256())) + log.Debug("Compression complete", "hash", hex.EncodeToString(writer.Sum256())) return nil } @@ -541,34 +578,42 @@ func (sm *SnapshotManager) compressFile(inputPath, outputPath string) error { // copyFile copies a file from src to dst func (sm *SnapshotManager) copyFile(src, dst string) error { log.Debug("Opening source file for copy", "path", src) + sourceFile, err := sm.fs.Open(src) if err != nil { return err } defer func() { log.Debug("Closing source file", "path", src) - if err := sourceFile.Close(); err != nil { + + err := sourceFile.Close() + if err != nil { log.Debug("Failed to close source file", "path", src, "error", err) } }() log.Debug("Creating destination file", "path", dst) + destFile, err := sm.fs.Create(dst) if err != nil { return err } defer func() { log.Debug("Closing destination file", "path", dst) - if err := destFile.Close(); err != nil { + + err := destFile.Close() + if err != nil { log.Debug("Failed to close destination file", "path", dst, "error", err) } }() log.Debug("Copying file data") + n, err := io.Copy(destFile, sourceFile) if err != nil { return err } + log.Debug("File copy complete", "bytes_copied", n) return nil @@ -576,7 +621,6 @@ func (sm *SnapshotManager) copyFile(src, dst string) error { // generateBlobManifest creates a compressed JSON list of all blobs in the snapshot func (sm *SnapshotManager) generateBlobManifest(ctx context.Context, dbPath string, snapshotID string) ([]byte, error) { - // Open the cleaned database using the database package db, err := database.New(ctx, dbPath) if err != nil { @@ -589,10 +633,12 @@ func (sm *SnapshotManager) generateBlobManifest(ctx context.Context, dbPath stri // Get all blobs for this snapshot log.Debug("Querying blobs for snapshot", "snapshot_id", snapshotID) + blobHashes, err := repos.Snapshots.GetBlobHashes(ctx, snapshotID) if err != nil { return nil, fmt.Errorf("getting snapshot blobs: %w", err) } + log.Debug("Found blobs", "count", len(blobHashes)) // Get blob details including sizes @@ -603,8 +649,10 @@ func (sm *SnapshotManager) generateBlobManifest(ctx context.Context, dbPath stri blob, err := repos.Blobs.GetByHash(ctx, hash) if err != nil { log.Warn("Failed to get blob details", "hash", hash, "error", err) + continue } + if blob != nil { blobs = append(blobs, BlobInfo{ Hash: hash, @@ -648,6 +696,7 @@ func (sm *SnapshotManager) getFileSize(path string) int64 { if err != nil { return -1 } + return info.Size() } @@ -663,6 +712,7 @@ type BackupStats struct { // ExtendedBackupStats contains additional statistics for comprehensive tracking type ExtendedBackupStats struct { BackupStats + BlobUncompressedSize int64 // Total uncompressed size of all referenced blobs CompressionLevel int // Compression level used for this snapshot UploadDurationMs int64 // Total milliseconds spent uploading to S3 @@ -682,6 +732,7 @@ func (sm *SnapshotManager) CleanupIncompleteSnapshots(ctx context.Context, hostn if len(incompleteSnapshots) == 0 { log.Debug("No incomplete snapshots found") + return nil } @@ -692,14 +743,15 @@ func (sm *SnapshotManager) CleanupIncompleteSnapshots(ctx context.Context, hostn // Check if metadata exists in storage (paths use the hashed // remote key so we don't leak host info to the listing). metadataKey := fmt.Sprintf("metadata/%s/db.zst", RemoteSnapshotKey(snapshot.ID.String())) - _, err := sm.storage.Stat(ctx, metadataKey) + _, err := sm.storage.Stat(ctx, metadataKey) if err != nil { // Metadata doesn't exist in S3 - this is an incomplete snapshot log.Info("Cleaning up incomplete snapshot record", "snapshot_id", snapshot.ID, "started_at", snapshot.StartedAt) // Delete the snapshot and all its associations - if err := sm.deleteSnapshot(ctx, snapshot.ID.String()); err != nil { + err := sm.deleteSnapshot(ctx, snapshot.ID.String()) + if err != nil { return fmt.Errorf("deleting incomplete snapshot %s: %w", snapshot.ID, err) } @@ -708,7 +760,9 @@ func (sm *SnapshotManager) CleanupIncompleteSnapshots(ctx context.Context, hostn // Metadata exists - this snapshot was completed but database wasn't updated // This shouldn't happen in normal operation, but mark it complete log.Warn("Found snapshot with remote metadata but incomplete in database", "snapshot_id", snapshot.ID) - if err := sm.repos.Snapshots.MarkComplete(ctx, nil, snapshot.ID.String()); err != nil { + + err := sm.repos.Snapshots.MarkComplete(ctx, nil, snapshot.ID.String()) + if err != nil { log.Error("Failed to mark snapshot as complete in database", "snapshot_id", snapshot.ID, "error", err) } } @@ -720,28 +774,34 @@ func (sm *SnapshotManager) CleanupIncompleteSnapshots(ctx context.Context, hostn // deleteSnapshot removes a snapshot and all its associations from the database func (sm *SnapshotManager) deleteSnapshot(ctx context.Context, snapshotID string) error { // Delete snapshot_files entries - if err := sm.repos.Snapshots.DeleteSnapshotFiles(ctx, snapshotID); err != nil { + err := sm.repos.Snapshots.DeleteSnapshotFiles(ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot files: %w", err) } // Delete snapshot_blobs entries - if err := sm.repos.Snapshots.DeleteSnapshotBlobs(ctx, snapshotID); err != nil { + err = sm.repos.Snapshots.DeleteSnapshotBlobs(ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot blobs: %w", err) } // Delete uploads entries (has foreign key to snapshots without CASCADE) - if err := sm.repos.Snapshots.DeleteSnapshotUploads(ctx, snapshotID); err != nil { + err = sm.repos.Snapshots.DeleteSnapshotUploads(ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot uploads: %w", err) } // Delete the snapshot itself - if err := sm.repos.Snapshots.Delete(ctx, snapshotID); err != nil { + err = sm.repos.Snapshots.Delete(ctx, snapshotID) + if err != nil { return fmt.Errorf("deleting snapshot: %w", err) } // Clean up orphaned data log.Debug("Cleaning up orphaned records in main database") - if err := sm.CleanupOrphanedData(ctx); err != nil { + + err = sm.CleanupOrphanedData(ctx) + if err != nil { return fmt.Errorf("cleaning up orphaned data: %w", err) } @@ -759,28 +819,36 @@ func (sm *SnapshotManager) CleanupOrphanedData(ctx context.Context) error { // Delete orphaned files (files not in any snapshot) log.Debug("Deleting orphaned file records from database") - if err := sm.repos.Files.DeleteOrphaned(ctx); err != nil { + + err := sm.repos.Files.DeleteOrphaned(ctx) + if err != nil { return fmt.Errorf("deleting orphaned files: %w", err) } // Delete orphaned blobs (blobs not in any snapshot) // This will cascade delete blob_chunks for deleted blobs log.Debug("Deleting orphaned blob records from database") - if err := sm.repos.Blobs.DeleteOrphaned(ctx); err != nil { + + err = sm.repos.Blobs.DeleteOrphaned(ctx) + if err != nil { return fmt.Errorf("deleting orphaned blobs: %w", err) } // Delete orphaned blob_chunks entries // This handles cases where the blob still exists but chunks were deleted log.Debug("Deleting orphaned blob_chunks associations from database") - if err := sm.repos.BlobChunks.DeleteOrphaned(ctx); err != nil { + + err = sm.repos.BlobChunks.DeleteOrphaned(ctx) + if err != nil { return fmt.Errorf("deleting orphaned blob_chunks: %w", err) } // Delete orphaned chunks (chunks not referenced by any file) // This must come after cleaning up blob_chunks to avoid foreign key violations log.Debug("Deleting orphaned chunk records from database") - if err := sm.repos.Chunks.DeleteOrphaned(ctx); err != nil { + + err = sm.repos.Chunks.DeleteOrphaned(ctx) + if err != nil { return fmt.Errorf("deleting orphaned chunks: %w", err) } @@ -793,21 +861,26 @@ func (sm *SnapshotManager) deleteOtherSnapshots(ctx context.Context, tx *sql.Tx, // First delete uploads that reference other snapshots (no CASCADE DELETE on this FK) database.LogSQL("Execute", "DELETE FROM uploads WHERE snapshot_id != ?", currentSnapshotID) + uploadResult, err := tx.ExecContext(ctx, "DELETE FROM uploads WHERE snapshot_id != ?", currentSnapshotID) if err != nil { return fmt.Errorf("deleting uploads for other snapshots: %w", err) } + uploadsDeleted, _ := uploadResult.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted upload records", "count", uploadsDeleted) // Now we can safely delete the snapshots database.LogSQL("Execute", "DELETE FROM snapshots WHERE id != ?", currentSnapshotID) + result, err := tx.ExecContext(ctx, "DELETE FROM snapshots WHERE id != ?", currentSnapshotID) if err != nil { return fmt.Errorf("deleting other snapshots: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted snapshot records from database", "count", rowsAffected) + return nil } @@ -816,22 +889,27 @@ func (sm *SnapshotManager) deleteOrphanedSnapshotAssociations(ctx context.Contex // Delete orphaned snapshot_files log.Debug("[Temp DB Cleanup] Deleting orphaned snapshot_files associations") database.LogSQL("Execute", "DELETE FROM snapshot_files WHERE snapshot_id != ?", currentSnapshotID) + result, err := tx.ExecContext(ctx, "DELETE FROM snapshot_files WHERE snapshot_id != ?", currentSnapshotID) if err != nil { return fmt.Errorf("deleting orphaned snapshot_files: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted snapshot_files associations", "count", rowsAffected) // Delete orphaned snapshot_blobs log.Debug("[Temp DB Cleanup] Deleting orphaned snapshot_blobs associations") database.LogSQL("Execute", "DELETE FROM snapshot_blobs WHERE snapshot_id != ?", currentSnapshotID) + result, err = tx.ExecContext(ctx, "DELETE FROM snapshot_blobs WHERE snapshot_id != ?", currentSnapshotID) if err != nil { return fmt.Errorf("deleting orphaned snapshot_blobs: %w", err) } + rowsAffected, _ = result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted snapshot_blobs associations", "count", rowsAffected) + return nil } @@ -839,6 +917,7 @@ func (sm *SnapshotManager) deleteOrphanedSnapshotAssociations(ctx context.Contex func (sm *SnapshotManager) deleteOrphanedFiles(ctx context.Context, tx *sql.Tx, currentSnapshotID string) error { log.Debug("[Temp DB Cleanup] Deleting file records not referenced by current snapshot") database.LogSQL("Execute", `DELETE FROM files WHERE NOT EXISTS (SELECT 1 FROM snapshot_files WHERE snapshot_files.file_id = files.id AND snapshot_files.snapshot_id = ?)`, currentSnapshotID) + result, err := tx.ExecContext(ctx, ` DELETE FROM files WHERE NOT EXISTS ( @@ -849,11 +928,13 @@ func (sm *SnapshotManager) deleteOrphanedFiles(ctx context.Context, tx *sql.Tx, if err != nil { return fmt.Errorf("deleting orphaned files: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted file records from database", "count", rowsAffected) // Note: file_chunks will be deleted via CASCADE log.Debug("[Temp DB Cleanup] file_chunks associations deleted via CASCADE") + return nil } @@ -861,6 +942,7 @@ func (sm *SnapshotManager) deleteOrphanedFiles(ctx context.Context, tx *sql.Tx, func (sm *SnapshotManager) deleteOrphanedChunkToFileMappings(ctx context.Context, tx *sql.Tx) error { log.Debug("[Temp DB Cleanup] Deleting orphaned chunk_files associations") database.LogSQL("Execute", `DELETE FROM chunk_files WHERE NOT EXISTS (SELECT 1 FROM files WHERE files.id = chunk_files.file_id)`) + result, err := tx.ExecContext(ctx, ` DELETE FROM chunk_files WHERE NOT EXISTS ( @@ -870,8 +952,10 @@ func (sm *SnapshotManager) deleteOrphanedChunkToFileMappings(ctx context.Context if err != nil { return fmt.Errorf("deleting orphaned chunk_files: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted chunk_files associations", "count", rowsAffected) + return nil } @@ -879,6 +963,7 @@ func (sm *SnapshotManager) deleteOrphanedChunkToFileMappings(ctx context.Context func (sm *SnapshotManager) deleteOrphanedBlobs(ctx context.Context, tx *sql.Tx, currentSnapshotID string) error { log.Debug("[Temp DB Cleanup] Deleting blob records not referenced by current snapshot") database.LogSQL("Execute", `DELETE FROM blobs WHERE NOT EXISTS (SELECT 1 FROM snapshot_blobs WHERE snapshot_blobs.blob_hash = blobs.blob_hash AND snapshot_blobs.snapshot_id = ?)`, currentSnapshotID) + result, err := tx.ExecContext(ctx, ` DELETE FROM blobs WHERE NOT EXISTS ( @@ -889,8 +974,10 @@ func (sm *SnapshotManager) deleteOrphanedBlobs(ctx context.Context, tx *sql.Tx, if err != nil { return fmt.Errorf("deleting orphaned blobs: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted blob records from database", "count", rowsAffected) + return nil } @@ -898,6 +985,7 @@ func (sm *SnapshotManager) deleteOrphanedBlobs(ctx context.Context, tx *sql.Tx, func (sm *SnapshotManager) deleteOrphanedBlobToChunkMappings(ctx context.Context, tx *sql.Tx) error { log.Debug("[Temp DB Cleanup] Deleting orphaned blob_chunks associations") database.LogSQL("Execute", `DELETE FROM blob_chunks WHERE NOT EXISTS (SELECT 1 FROM blobs WHERE blobs.id = blob_chunks.blob_id)`) + result, err := tx.ExecContext(ctx, ` DELETE FROM blob_chunks WHERE NOT EXISTS ( @@ -907,14 +995,17 @@ func (sm *SnapshotManager) deleteOrphanedBlobToChunkMappings(ctx context.Context if err != nil { return fmt.Errorf("deleting orphaned blob_chunks: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted blob_chunks associations", "count", rowsAffected) + return nil } // deleteOrphanedChunks deletes chunks not referenced by any file or blob func (sm *SnapshotManager) deleteOrphanedChunks(ctx context.Context, tx *sql.Tx) error { log.Debug("[Temp DB Cleanup] Deleting orphaned chunk records") + query := ` DELETE FROM chunks WHERE NOT EXISTS ( @@ -926,11 +1017,14 @@ func (sm *SnapshotManager) deleteOrphanedChunks(ctx context.Context, tx *sql.Tx) WHERE blob_chunks.chunk_hash = chunks.chunk_hash )` database.LogSQL("Execute", query) + result, err := tx.ExecContext(ctx, query) if err != nil { return fmt.Errorf("deleting orphaned chunks: %w", err) } + rowsAffected, _ := result.RowsAffected() log.Debug("[Temp DB Cleanup] Deleted chunk records from database", "count", rowsAffected) + return nil } diff --git a/internal/snapshot/snapshot_test.go b/internal/snapshot/snapshot_test.go index a47df4a..969aeed 100644 --- a/internal/snapshot/snapshot_test.go +++ b/internal/snapshot/snapshot_test.go @@ -33,6 +33,7 @@ func copyFile(fs afero.Fs, src, dst string) error { defer func() { _ = destFile.Close() }() _, err = io.Copy(destFile, sourceFile) + return err } @@ -46,6 +47,7 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { // Create a test database tempDir := t.TempDir() dbPath := filepath.Join(tempDir, "test.db") + db, err := database.New(ctx, dbPath) if err != nil { t.Fatalf("failed to create database: %v", err) @@ -71,9 +73,11 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { chunk := &database.Chunk{ChunkHash: "orphan-chunk", Size: 500} err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error { - if err := repos.Files.Create(ctx, tx, file); err != nil { + err := repos.Files.Create(ctx, tx, file) + if err != nil { return err } + return repos.Chunks.Create(ctx, tx, chunk) }) if err != nil { @@ -111,7 +115,8 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { t.Fatalf("failed to open cleaned database: %v", err) } defer func() { - if err := cleanedDB.Close(); err != nil { + err := cleanedDB.Close() + if err != nil { t.Errorf("failed to close database: %v", err) } }() @@ -123,6 +128,7 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { if err != nil { t.Fatalf("failed to get snapshot: %v", err) } + if verifySnapshot == nil { t.Error("snapshot should exist") } @@ -132,6 +138,7 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { if err != nil { t.Fatalf("failed to check file: %v", err) } + if f != nil { t.Error("orphan file should not exist") } @@ -141,6 +148,7 @@ func TestCleanSnapshotDBEmptySnapshot(t *testing.T) { if err != nil { t.Fatalf("failed to check chunk: %v", err) } + if c != nil { t.Error("orphan chunk should not exist") } @@ -156,6 +164,7 @@ func TestCleanSnapshotDBNonExistentSnapshot(t *testing.T) { // Create a test database tempDir := t.TempDir() dbPath := filepath.Join(tempDir, "test.db") + db, err := database.New(ctx, dbPath) if err != nil { t.Fatalf("failed to create database: %v", err)