Apply mechanical lint fixes for golangci-lint v2.12.2 rollout
Auto-remediate style-only findings (wsl_v5, nlreturn, noinlineerr, modernize, intrange, perfsprint, usetesting, unconvert, errorlint, gocritic, testifylint) and rename printf-style helpers to f-suffixed names (goprintffuncname): ui.Writer message methods, cli.ReportErrorf, database.Fatalf, vaultik stdoutf.
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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[:])
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -213,8 +219,10 @@ 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 {
|
||||
s.ui.Beginf("Associating %s unchanged files with the snapshot.", s.ui.Count(len(scanResult.UnchangedFileIDs)))
|
||||
|
||||
err := s.batchAddFilesToSnapshot(ctx, scanResult.UnchangedFileIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("associating unchanged files: %w", err)
|
||||
}
|
||||
}
|
||||
@@ -224,13 +232,15 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
|
||||
|
||||
// Phase 2: Process files and create chunks
|
||||
if len(filesToProcess) > 0 {
|
||||
s.ui.Begin("Backing up %s snapshot source files (chunking, compressing, encrypting, uploading).", s.ui.Count(len(filesToProcess)))
|
||||
s.ui.Beginf("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 {
|
||||
s.ui.Info("Snapshot file backup skipped: no changed files (creating metadata-only snapshot).")
|
||||
s.ui.Infof("Snapshot file backup skipped: no changed files (creating metadata-only snapshot).")
|
||||
log.Info("Phase 2/3: Skipping (no files need processing, metadata-only snapshot)")
|
||||
}
|
||||
|
||||
@@ -243,18 +253,22 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
|
||||
// loadDatabaseState loads known files and chunks from the database into memory for fast lookup
|
||||
// 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.")
|
||||
s.ui.Beginf("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.")
|
||||
s.ui.Completef("Loaded %s known files from local index database.", s.ui.Count(len(knownFiles)))
|
||||
|
||||
s.ui.Beginf("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)))
|
||||
|
||||
s.ui.Completef("Loaded %s known chunks from local index database.", s.ui.Count(len(s.knownChunks)))
|
||||
|
||||
return knownFiles, nil
|
||||
}
|
||||
@@ -288,7 +302,8 @@ func (s *Scanner) summarizeScanPhase(result *ScanResult, filesToProcess []*FileT
|
||||
s.ui.Count(result.FilesDeleted),
|
||||
s.ui.Size(result.BytesDeleted))
|
||||
}
|
||||
s.ui.Complete("%s.", msg)
|
||||
|
||||
s.ui.Completef("%s.", msg)
|
||||
}
|
||||
|
||||
// finalizeScanResult populates final blob statistics in the scan result
|
||||
@@ -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)
|
||||
s.ui.Errorf("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,16 +852,16 @@ 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.Progressf("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)),
|
||||
s.ui.Percent(pct),
|
||||
s.ui.Count(changedCount),
|
||||
@@ -786,7 +870,7 @@ func (s *Scanner) printScanProgressLine(filesScanned int64, changedCount int, es
|
||||
s.ui.Time(time.Now().Add(eta)),
|
||||
s.ui.Duration(eta))
|
||||
} else {
|
||||
s.ui.Progress("Snapshot source files enumeration: %s files (~%s), %s changed or new, %.0f files/sec, enumeration elapsed: %s.",
|
||||
s.ui.Progressf("Snapshot source files enumeration: %s files (~%s), %s changed or new, %.0f files/sec, enumeration elapsed: %s.",
|
||||
s.ui.Count(int(filesScanned)),
|
||||
s.ui.Percent(pct),
|
||||
s.ui.Count(changedCount),
|
||||
@@ -794,7 +878,7 @@ func (s *Scanner) printScanProgressLine(filesScanned int64, changedCount int, es
|
||||
s.ui.Duration(elapsed))
|
||||
}
|
||||
} else {
|
||||
s.ui.Progress("Snapshot source files enumeration: %s files seen, %s changed or new, %.0f files/sec, enumeration elapsed: %s.",
|
||||
s.ui.Progressf("Snapshot source files enumeration: %s files seen, %s changed or new, %.0f files/sec, enumeration elapsed: %s.",
|
||||
s.ui.Count(int(filesScanned)),
|
||||
s.ui.Count(changedCount),
|
||||
rate,
|
||||
@@ -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 {
|
||||
@@ -964,15 +1051,16 @@ func (s *Scanner) batchAddFilesToSnapshot(ctx context.Context, fileIDs []types.F
|
||||
elapsed := time.Since(startTime)
|
||||
rate := float64(end) / elapsed.Seconds()
|
||||
pct := float64(end) / float64(len(fileIDs)) * 100
|
||||
s.ui.Progress("Snapshot unchanged-file association: %s/%s (%s), %.0f files/sec.",
|
||||
s.ui.Progressf("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()
|
||||
}
|
||||
}
|
||||
|
||||
elapsed := time.Since(startTime)
|
||||
rate := float64(len(fileIDs)) / elapsed.Seconds()
|
||||
s.ui.Complete("Associated %s unchanged files with the snapshot in %s (%.0f files/sec).",
|
||||
s.ui.Completef("Associated %s unchanged files with the snapshot in %s (%.0f files/sec).",
|
||||
s.ui.Count(len(fileIDs)), s.ui.Duration(elapsed), rate)
|
||||
|
||||
return nil
|
||||
@@ -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)
|
||||
s.ui.Errorf("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,13 +1160,14 @@ 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
|
||||
}
|
||||
|
||||
if eta > 0 {
|
||||
s.ui.Progress("Snapshot backup: %s/%s files (%s), %s/%s, %s, %.0f files/sec, backup elapsed: %s, backup ETA: %s (est remain %s).",
|
||||
s.ui.Progressf("Snapshot backup: %s/%s files (%s), %s/%s, %s, %.0f files/sec, backup elapsed: %s, backup ETA: %s (est remain %s).",
|
||||
s.ui.Count(filesProcessed),
|
||||
s.ui.Count(totalFiles),
|
||||
s.ui.Percent(pct),
|
||||
@@ -1079,7 +1179,7 @@ func (s *Scanner) printProcessingProgress(filesProcessed, totalFiles int, bytesP
|
||||
s.ui.Time(time.Now().Add(eta)),
|
||||
s.ui.Duration(eta))
|
||||
} else {
|
||||
s.ui.Progress("Snapshot backup: %s/%s files (%s), %s/%s, %s, %.0f files/sec, backup elapsed: %s.",
|
||||
s.ui.Progressf("Snapshot backup: %s/%s files (%s), %s/%s, %s, %.0f files/sec, backup elapsed: %s.",
|
||||
s.ui.Count(filesProcessed),
|
||||
s.ui.Count(totalFiles),
|
||||
s.ui.Percent(pct),
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -1181,25 +1290,28 @@ func (s *Scanner) uploadBlobIfNeeded(ctx context.Context, blobPath string, blobW
|
||||
if _, err := s.storage.Stat(ctx, blobPath); err == nil {
|
||||
log.Info("Blob already exists in storage, skipping upload",
|
||||
"hash", finishedBlob.Hash, "size", humanize.Bytes(uint64(finishedBlob.Compressed)))
|
||||
s.ui.Info("Blob %s (%s) already exists at %s. Skipping upload.",
|
||||
s.ui.Infof("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
|
||||
}
|
||||
|
||||
s.ui.Begin("Uploading blob %s (%s) to %s.",
|
||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed), s.ui.Path(destination))
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
uploadDuration := time.Since(startTime)
|
||||
uploadSpeedBps := float64(finishedBlob.Compressed) / uploadDuration.Seconds()
|
||||
|
||||
s.ui.Complete("Uploaded blob %s (%s) in %s at %s.",
|
||||
s.ui.Completef("Uploaded blob %s (%s) in %s at %s.",
|
||||
s.ui.Hex(finishedBlob.Hash),
|
||||
s.ui.Size(finishedBlob.Compressed),
|
||||
s.ui.Duration(uploadDuration),
|
||||
@@ -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,11 +1364,13 @@ 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.Progressf("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),
|
||||
s.ui.Size(finishedBlob.Compressed),
|
||||
@@ -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{
|
||||
@@ -1489,7 +1629,7 @@ func (s *Scanner) detectDeletedFilesFromMap(ctx context.Context, knownFiles map[
|
||||
}
|
||||
|
||||
if result.FilesDeleted > 0 {
|
||||
s.ui.Info("Snapshot source files enumeration detected %s deleted files.", s.ui.Count(result.FilesDeleted))
|
||||
s.ui.Infof("Snapshot source files enumeration detected %s deleted files.", s.ui.Count(result.FilesDeleted))
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user