sneak/vaultik
1
0
Fork 0
Code Issues 19 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Apply linter autofixes: internal/snapshot (refs #61)

Browse Source
This commit is contained in:
Jeffrey Paul
2026-08-07 16:53:23 +00:00
parent bec964fc20
commit 1e05fa0dd7
11 changed files with 452 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
internal/snapshot/scanner.go
View File

@@ -119,6 +119,7 @@ func NewScanner(cfg ScannerConfig) *Scanner {
// Create encryptor (required for blob packing)
if len(cfg.AgeRecipients) == 0 {
log.Error("No age recipients configured - encryption is required")
return nil
}
@@ -130,9 +131,11 @@ func NewScanner(cfg ScannerConfig) *Scanner {
Repositories: cfg.Repositories,
Fs: cfg.FS,
}
packer, err := blob.NewPacker(packerCfg)
if err != nil {
log.Error("Failed to create packer", "error", err)
return nil
}
@@ -199,11 +202,14 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
// Phase 1: Scan directory, collect files to process, and track existing files
// (builds existingFiles map during walk to avoid double traversal)
log.Info("Phase 1/3: Scanning directory structure")
existingFiles := make(map[string]struct{})
scanResult, err := s.scanPhase(ctx, path, result, existingFiles, knownFiles)
if err != nil {
return nil, fmt.Errorf("scan phase failed: %w", err)
}
filesToProcess := scanResult.FilesToProcess
// Phase 1b: Detect deleted files by comparing DB against scanned files
@@ -214,7 +220,9 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
// Phase 1c: Associate unchanged files with this snapshot (no new records needed)
if len(scanResult.UnchangedFileIDs) > 0 {
s.ui.Begin("Associating %s unchanged files with the snapshot.", s.ui.Count(len(scanResult.UnchangedFileIDs)))
if err := s.batchAddFilesToSnapshot(ctx, scanResult.UnchangedFileIDs); err != nil {
err := s.batchAddFilesToSnapshot(ctx, scanResult.UnchangedFileIDs)
if err != nil {
return nil, fmt.Errorf("associating unchanged files: %w", err)
}
}
@@ -226,7 +234,9 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
if len(filesToProcess) > 0 {
s.ui.Begin("Backing up %s snapshot source files (chunking, compressing, encrypting, uploading).", s.ui.Count(len(filesToProcess)))
log.Info("Phase 2/3: Creating snapshot (chunking, compressing, encrypting, and uploading blobs)")
if err := s.processPhase(ctx, filesToProcess, result); err != nil {
err := s.processPhase(ctx, filesToProcess, result)
if err != nil {
return nil, fmt.Errorf("process phase failed: %w", err)
}
} else {
@@ -244,16 +254,20 @@ func (s *Scanner) Scan(ctx context.Context, path string, snapshotID string) (*Sc
// This avoids per-file and per-chunk database queries during the scan and process phases
func (s *Scanner) loadDatabaseState(ctx context.Context, path string) (map[string]*database.File, error) {
s.ui.Begin("Loading known files from local index database.")
knownFiles, err := s.loadKnownFiles(ctx, path)
if err != nil {
return nil, fmt.Errorf("loading known files: %w", err)
}
s.ui.Complete("Loaded %s known files from local index database.", s.ui.Count(len(knownFiles)))
s.ui.Begin("Loading known chunks from local index database.")
if err := s.loadKnownChunks(ctx); err != nil {
return nil, fmt.Errorf("loading known chunks: %w", err)
}
s.ui.Complete("Loaded %s known chunks from local index database.", s.ui.Count(len(s.knownChunks)))
return knownFiles, nil
@@ -288,6 +302,7 @@ func (s *Scanner) summarizeScanPhase(result *ScanResult, filesToProcess []*FileT
s.ui.Count(result.FilesDeleted),
s.ui.Size(result.BytesDeleted))
}
s.ui.Complete("%s.", msg)
}
@@ -337,6 +352,7 @@ func (s *Scanner) loadKnownChunks(ctx context.Context) error {
}
s.knownChunksMu.Lock()
s.knownChunks = make(map[string]struct{}, len(chunks))
for _, c := range chunks {
s.knownChunks[c.ChunkHash.String()] = struct{}{}
@@ -351,6 +367,7 @@ func (s *Scanner) chunkExists(hash string) bool {
s.knownChunksMu.RLock()
_, exists := s.knownChunks[hash]
s.knownChunksMu.RUnlock()
return exists
}
@@ -371,7 +388,9 @@ func (s *Scanner) addPendingChunkHash(hash string) {
// removePendingChunkHashes removes committed chunk hashes from the pending set
func (s *Scanner) removePendingChunkHashes(hashes []string) {
log.Debug("removePendingChunkHashes: starting", "count", len(hashes))
start := time.Now()
s.pendingChunkHashesMu.Lock()
for _, hash := range hashes {
delete(s.pendingChunkHashes, hash)
@@ -385,6 +404,7 @@ func (s *Scanner) isChunkPending(hash string) bool {
s.pendingChunkHashesMu.Lock()
_, pending := s.pendingChunkHashes[hash]
s.pendingChunkHashesMu.Unlock()
return pending
}
@@ -411,37 +431,45 @@ func (s *Scanner) flushPendingFiles(ctx context.Context) error {
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
for _, data := range files {
// Create or update the file record
if err := s.repos.Files.Create(txCtx, tx, data.file); err != nil {
err := s.repos.Files.Create(txCtx, tx, data.file)
if err != nil {
return fmt.Errorf("creating file record: %w", err)
}
// Delete any existing file_chunks and chunk_files for this file
if err := s.repos.FileChunks.DeleteByFileID(txCtx, tx, data.file.ID); err != nil {
err = s.repos.FileChunks.DeleteByFileID(txCtx, tx, data.file.ID)
if err != nil {
return fmt.Errorf("deleting old file chunks: %w", err)
}
if err := s.repos.ChunkFiles.DeleteByFileID(txCtx, tx, data.file.ID); err != nil {
err = s.repos.ChunkFiles.DeleteByFileID(txCtx, tx, data.file.ID)
if err != nil {
return fmt.Errorf("deleting old chunk files: %w", err)
}
// Create file-chunk mappings
for i := range data.fileChunks {
if err := s.repos.FileChunks.Create(txCtx, tx, &data.fileChunks[i]); err != nil {
err := s.repos.FileChunks.Create(txCtx, tx, &data.fileChunks[i])
if err != nil {
return fmt.Errorf("creating file chunk: %w", err)
}
}
// Create chunk-file mappings
for i := range data.chunkFiles {
if err := s.repos.ChunkFiles.Create(txCtx, tx, &data.chunkFiles[i]); err != nil {
err := s.repos.ChunkFiles.Create(txCtx, tx, &data.chunkFiles[i])
if err != nil {
return fmt.Errorf("creating chunk file: %w", err)
}
}
// Add file to snapshot
if err := s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, data.file.ID); err != nil {
err = s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, data.file.ID)
if err != nil {
return fmt.Errorf("adding file to snapshot: %w", err)
}
}
return nil
})
}
@@ -455,6 +483,7 @@ func (s *Scanner) flushAllPending(ctx context.Context) error {
// Files with pending chunks are kept in the queue for later flushing
func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error {
flushStart := time.Now()
log.Debug("flushCompletedPendingFiles: starting")
// Partition pending files into those ready to flush and those still waiting
@@ -462,6 +491,7 @@ func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error {
if len(canFlush) == 0 {
log.Debug("flushCompletedPendingFiles: nothing to flush")
return nil
}
@@ -474,10 +504,13 @@ func (s *Scanner) flushCompletedPendingFiles(ctx context.Context) error {
// Execute the batch flush in a single transaction
log.Debug("flushCompletedPendingFiles: starting transaction")
txStart := time.Now()
err := s.executeBatchFileFlush(ctx, allFiles, allFileIDs, allFileChunks, allChunkFiles)
log.Debug("flushCompletedPendingFiles: transaction done", "duration", time.Since(txStart))
log.Debug("flushCompletedPendingFiles: total duration", "duration", time.Since(flushStart))
return err
}
@@ -492,21 +525,27 @@ func (s *Scanner) partitionPendingByChunkStatus() (canFlush []pendingFileData, s
var stillPending []pendingFileData
log.Debug("flushCompletedPendingFiles: checking which files can flush")
checkStart := time.Now()
for _, data := range s.pendingFiles {
allChunksCommitted := true
for _, fc := range data.fileChunks {
if s.isChunkPending(fc.ChunkHash.String()) {
allChunksCommitted = false
break
}
}
if allChunksCommitted {
canFlush = append(canFlush, data)
} else {
stillPending = append(stillPending, data)
}
}
log.Debug("flushCompletedPendingFiles: check done", "duration", time.Since(checkStart), "can_flush", len(canFlush), "still_pending", len(stillPending))
s.pendingFiles = stillPending
@@ -520,12 +559,15 @@ func (s *Scanner) partitionPendingByChunkStatus() (canFlush []pendingFileData, s
// mappings from the given pending file data for efficient batch database operations
func (s *Scanner) collectBatchFlushData(canFlush []pendingFileData) ([]*database.File, []types.FileID, []database.FileChunk, []database.ChunkFile) {
log.Debug("flushCompletedPendingFiles: collecting data for batch ops")
collectStart := time.Now()
var allFileChunks []database.FileChunk
var allChunkFiles []database.ChunkFile
var allFileIDs []types.FileID
var allFiles []*database.File
var (
allFileChunks []database.FileChunk
allChunkFiles []database.ChunkFile
allFileIDs []types.FileID
allFiles []*database.File
)
for _, data := range canFlush {
allFileChunks = append(allFileChunks, data.fileChunks...)
@@ -551,52 +593,77 @@ func (s *Scanner) executeBatchFileFlush(ctx context.Context, allFiles []*databas
// Batch delete old file_chunks and chunk_files
log.Debug("flushCompletedPendingFiles: deleting old file_chunks")
opStart := time.Now()
if err := s.repos.FileChunks.DeleteByFileIDs(txCtx, tx, allFileIDs); err != nil {
err := s.repos.FileChunks.DeleteByFileIDs(txCtx, tx, allFileIDs)
if err != nil {
return fmt.Errorf("batch deleting old file chunks: %w", err)
}
log.Debug("flushCompletedPendingFiles: deleted file_chunks", "duration", time.Since(opStart))
log.Debug("flushCompletedPendingFiles: deleting old chunk_files")
opStart = time.Now()
if err := s.repos.ChunkFiles.DeleteByFileIDs(txCtx, tx, allFileIDs); err != nil {
err = s.repos.ChunkFiles.DeleteByFileIDs(txCtx, tx, allFileIDs)
if err != nil {
return fmt.Errorf("batch deleting old chunk files: %w", err)
}
log.Debug("flushCompletedPendingFiles: deleted chunk_files", "duration", time.Since(opStart))
// Batch create/update file records
log.Debug("flushCompletedPendingFiles: creating files")
opStart = time.Now()
if err := s.repos.Files.CreateBatch(txCtx, tx, allFiles); err != nil {
err = s.repos.Files.CreateBatch(txCtx, tx, allFiles)
if err != nil {
return fmt.Errorf("batch creating file records: %w", err)
}
log.Debug("flushCompletedPendingFiles: created files", "duration", time.Since(opStart))
// Batch insert file_chunks
log.Debug("flushCompletedPendingFiles: inserting file_chunks")
opStart = time.Now()
if err := s.repos.FileChunks.CreateBatch(txCtx, tx, allFileChunks); err != nil {
err = s.repos.FileChunks.CreateBatch(txCtx, tx, allFileChunks)
if err != nil {
return fmt.Errorf("batch creating file chunks: %w", err)
}
log.Debug("flushCompletedPendingFiles: inserted file_chunks", "duration", time.Since(opStart))
// Batch insert chunk_files
log.Debug("flushCompletedPendingFiles: inserting chunk_files")
opStart = time.Now()
if err := s.repos.ChunkFiles.CreateBatch(txCtx, tx, allChunkFiles); err != nil {
err = s.repos.ChunkFiles.CreateBatch(txCtx, tx, allChunkFiles)
if err != nil {
return fmt.Errorf("batch creating chunk files: %w", err)
}
log.Debug("flushCompletedPendingFiles: inserted chunk_files", "duration", time.Since(opStart))
// Batch add files to snapshot
log.Debug("flushCompletedPendingFiles: adding files to snapshot")
opStart = time.Now()
if err := s.repos.Snapshots.AddFilesByIDBatch(txCtx, tx, s.snapshotID, allFileIDs); err != nil {
err = s.repos.Snapshots.AddFilesByIDBatch(txCtx, tx, s.snapshotID, allFileIDs)
if err != nil {
return fmt.Errorf("batch adding files to snapshot: %w", err)
}
log.Debug("flushCompletedPendingFiles: added files to snapshot", "duration", time.Since(opStart))
log.Debug("flushCompletedPendingFiles: transaction complete")
return nil
})
}
@@ -616,24 +683,31 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
estimatedTotal := int64(len(knownFiles))
var filesToProcess []*FileToProcess
var unchangedFileIDs []types.FileID // Just IDs - no new records needed
var mu sync.Mutex
// Set up periodic status output
startTime := time.Now()
lastStatusTime := time.Now()
statusInterval := 15 * time.Second
var filesScanned int64
log.Debug("Starting directory walk", "path", path)
err := afero.Walk(s.fs, path, func(filePath string, info os.FileInfo, err error) error {
if err != nil {
if s.skipErrors {
log.Error("Failed to access file (skipping due to --skip-errors)", "path", filePath, "error", err)
s.ui.Error("Failed to access %s: %v. Skipping (--skip-errors).", s.ui.Path(filePath), err)
return nil // Continue scanning
}
log.Debug("Error accessing filesystem entry", "path", filePath, "error", err)
return wrapPermissionError(filePath, err)
}
@@ -649,6 +723,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
if info.IsDir() {
return filepath.SkipDir
}
return nil
}
@@ -657,7 +732,9 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
file := s.buildSymlinkEntry(filePath, info)
if file != nil {
existingFiles[filePath] = struct{}{}
mu.Lock()
filesToProcess = append(filesToProcess, &FileToProcess{
Path: filePath,
FileInfo: info,
@@ -667,6 +744,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
mu.Unlock()
s.updateScanEntryStats(result, true, info)
}
return nil
}
@@ -674,7 +752,9 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
if info.IsDir() {
file := s.buildDirectoryEntry(filePath, info)
existingFiles[filePath] = struct{}{}
mu.Lock()
filesToProcess = append(filesToProcess, &FileToProcess{
Path: filePath,
FileInfo: info,
@@ -682,6 +762,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
})
filesScanned++
mu.Unlock()
return nil
}
@@ -708,6 +789,7 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
// Unchanged file with existing ID - just need snapshot association
unchangedFileIDs = append(unchangedFileIDs, file.ID)
}
filesScanned++
changedCount := len(filesToProcess)
mu.Unlock()
@@ -718,12 +800,12 @@ func (s *Scanner) scanPhase(ctx context.Context, path string, result *ScanResult
// Output periodic status
if time.Since(lastStatusTime) >= statusInterval {
s.printScanProgressLine(filesScanned, changedCount, estimatedTotal, startTime)
lastStatusTime = time.Now()
}
return nil
})
if err != nil {
return nil, err
}
@@ -745,11 +827,13 @@ func (s *Scanner) updateScanEntryStats(result *ScanResult, needsProcessing bool,
} else {
result.FilesSkipped++
result.BytesSkipped += info.Size()
if s.progress != nil {
s.progress.GetStats().FilesSkipped.Add(1)
s.progress.GetStats().BytesSkipped.Add(info.Size())
}
}
result.FilesScanned++
if s.progress != nil {
s.progress.GetStats().FilesScanned.Add(1)
@@ -768,14 +852,14 @@ func (s *Scanner) printScanProgressLine(filesScanned int64, changedCount int, es
if pct > 100 {
pct = 100 // Cap at 100% for display
}
remaining := estimatedTotal - filesScanned
if remaining < 0 {
remaining = 0
}
remaining := max(estimatedTotal-filesScanned, 0)
var eta time.Duration
if rate > 0 && remaining > 0 {
eta = time.Duration(float64(remaining)/rate) * time.Second
}
if eta > 0 {
s.ui.Progress("Snapshot source files enumeration: %s files (~%s), %s changed or new, %.0f files/sec, enumeration elapsed: %s, enumeration ETA: %s (est remain %s).",
s.ui.Count(int(filesScanned)),
@@ -808,6 +892,7 @@ func (s *Scanner) buildSymlinkEntry(path string, info os.FileInfo) *database.Fil
target, err := os.Readlink(path)
if err != nil {
log.Debug("Cannot read symlink target", "path", path, "error", err)
return nil
}
@@ -860,9 +945,11 @@ func (s *Scanner) buildDirectoryEntry(path string, info os.FileInfo) *database.F
// and associates it with the current snapshot. No chunking is performed.
func (s *Scanner) recordNonRegularFile(ctx context.Context, ftp *FileToProcess) error {
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
if err := s.repos.Files.Create(txCtx, tx, ftp.File); err != nil {
err := s.repos.Files.Create(txCtx, tx, ftp.File)
if err != nil {
return fmt.Errorf("creating non-regular file record: %w", err)
}
return s.repos.Snapshots.AddFileByID(txCtx, tx, s.snapshotID, ftp.File.ID)
})
}
@@ -941,18 +1028,18 @@ func (s *Scanner) batchAddFilesToSnapshot(ctx context.Context, fileIDs []types.F
default:
}
end := i + batchSize
if end > len(fileIDs) {
end = len(fileIDs)
}
end := min(i+batchSize, len(fileIDs))
batch := fileIDs[i:end]
err := s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
for _, fileID := range batch {
if err := s.repos.Snapshots.AddFileByID(ctx, tx, s.snapshotID, fileID); err != nil {
err := s.repos.Snapshots.AddFileByID(ctx, tx, s.snapshotID, fileID)
if err != nil {
return fmt.Errorf("adding file to snapshot: %w", err)
}
}
return nil
})
if err != nil {
@@ -966,6 +1053,7 @@ func (s *Scanner) batchAddFilesToSnapshot(ctx context.Context, fileIDs []types.F
pct := float64(end) / float64(len(fileIDs)) * 100
s.ui.Progress("Snapshot unchanged-file association: %s/%s (%s), %.0f files/sec.",
s.ui.Count(end), s.ui.Count(len(fileIDs)), s.ui.Percent(pct), rate)
lastStatusTime = time.Now()
}
}
@@ -991,7 +1079,9 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc
statusInterval := 15 * time.Second
startTime := time.Now()
filesProcessed := 0
var bytesProcessed int64
totalFiles := len(filesToProcess)
// Process each file
@@ -1006,6 +1096,7 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc
if err != nil {
return err
}
if skipped {
continue
}
@@ -1021,6 +1112,7 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc
// Output periodic status
if time.Since(lastStatusTime) >= statusInterval {
s.printProcessingProgress(filesProcessed, totalFiles, bytesProcessed, totalBytes, startTime)
lastStatusTime = time.Now()
}
}
@@ -1032,22 +1124,29 @@ func (s *Scanner) processPhase(ctx context.Context, filesToProcess []*FileToProc
// processFileWithErrorHandling wraps processFileStreaming with error recovery for
// deleted files and skip-errors mode. Returns (skipped, error).
func (s *Scanner) processFileWithErrorHandling(ctx context.Context, fileToProcess *FileToProcess, result *ScanResult) (bool, error) {
if err := s.processFileStreaming(ctx, fileToProcess, result); err != nil {
err := s.processFileStreaming(ctx, fileToProcess, result)
if err != nil {
// Handle files that were deleted between scan and process phases
if errors.Is(err, os.ErrNotExist) {
log.Warn("File was deleted during backup, skipping", "path", fileToProcess.Path)
result.FilesSkipped++
return true, nil
}
// Skip file read errors if --skip-errors is enabled
if s.skipErrors {
log.Error("Failed to process file (skipping due to --skip-errors)", "path", fileToProcess.Path, "error", err)
s.ui.Error("Failed to process %s: %v. Skipping (--skip-errors).", s.ui.Path(fileToProcess.Path), err)
result.FilesSkipped++
return true, nil
}
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
}
return false, nil
}
@@ -1061,6 +1160,7 @@ func (s *Scanner) printProcessingProgress(filesProcessed, totalFiles int, bytesP
// Calculate ETA based on bytes (more accurate than files)
remainingBytes := totalBytes - bytesProcessed
var eta time.Duration
if byteRate > 0 {
eta = time.Duration(float64(remainingBytes)/byteRate) * time.Second
@@ -1097,15 +1197,19 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
// Final packer flush first - this commits remaining chunks to DB
// and handleBlobReady will flush files whose chunks are now committed
s.packerMu.Lock()
if err := s.packer.Flush(); err != nil {
err := s.packer.Flush()
if err != nil {
s.packerMu.Unlock()
return fmt.Errorf("flushing packer: %w", err)
}
s.packerMu.Unlock()
// Flush any remaining pending files (e.g., files with only pre-existing chunks
// that didn't trigger a blob finalize)
if err := s.flushAllPending(ctx); err != nil {
err = s.flushAllPending(ctx)
if err != nil {
return fmt.Errorf("flushing remaining pending files: %w", err)
}
@@ -1119,6 +1223,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
if err != nil {
return fmt.Errorf("parsing blob ID: %w", err)
}
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
return s.repos.Snapshots.AddBlob(ctx, tx, s.snapshotID, blobID, types.BlobHash(b.Hash))
})
@@ -1126,6 +1231,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
return fmt.Errorf("storing blob metadata: %w", err)
}
}
result.BlobsCreated += len(blobs)
}
@@ -1148,14 +1254,17 @@ func (s *Scanner) handleBlobReady(blobWithReader *blob.BlobWithReader) error {
}
blobPath := fmt.Sprintf("blobs/%s/%s/%s", finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
blobExists, err := s.uploadBlobIfNeeded(ctx, blobPath, blobWithReader, startTime)
if err != nil {
s.cleanupBlobTempFile(blobWithReader)
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
}
if err := s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime); err != nil {
s.cleanupBlobTempFile(blobWithReader)
return err
}
@@ -1183,6 +1292,7 @@ func (s *Scanner) uploadBlobIfNeeded(ctx context.Context, blobPath string, blobW
"hash", finishedBlob.Hash, "size", humanize.Bytes(uint64(finishedBlob.Compressed)))
s.ui.Info("Blob %s (%s) already exists at %s. Skipping upload.",
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed), s.ui.Path(destination))
return true, nil
}
@@ -1191,8 +1301,10 @@ func (s *Scanner) uploadBlobIfNeeded(ctx context.Context, blobPath string, blobW
progressCallback := s.makeUploadProgressCallback(ctx, finishedBlob, startTime)
if err := s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader, finishedBlob.Compressed, progressCallback); err != nil {
err := s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader, finishedBlob.Compressed, progressCallback)
if err != nil {
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
return false, fmt.Errorf("uploading blob to storage: %w", err)
}
@@ -1228,17 +1340,21 @@ func (s *Scanner) makeUploadProgressCallback(ctx context.Context, finishedBlob *
lastProgressTime := time.Now()
lastProgressBytes := int64(0)
lastStdoutTime := time.Now()
const stdoutInterval = 15 * time.Second
return func(uploaded int64) error {
now := time.Now()
elapsed := now.Sub(lastProgressTime).Seconds()
if elapsed > 0.5 {
bytesSinceLastUpdate := uploaded - lastProgressBytes
speed := float64(bytesSinceLastUpdate) / elapsed
if s.progress != nil {
s.progress.ReportUploadProgress(finishedBlob.Hash, uploaded, finishedBlob.Compressed, speed)
}
lastProgressTime = now
lastProgressBytes = uploaded
}
@@ -1248,10 +1364,12 @@ func (s *Scanner) makeUploadProgressCallback(ctx context.Context, finishedBlob *
totalElapsed := now.Sub(uploadStart)
pct := float64(uploaded) / float64(finishedBlob.Compressed) * 100
avgSpeed := float64(uploaded) / totalElapsed.Seconds()
var eta time.Duration
if avgSpeed > 0 {
eta = time.Duration(float64(finishedBlob.Compressed-uploaded)/avgSpeed) * time.Second
}
s.ui.Progress("Blob upload %s: %s / %s (%s) at %s, blob upload elapsed: %s, blob upload ETA: %s (est remain %s).",
s.ui.Hex(finishedBlob.Hash),
s.ui.Size(uploaded),
@@ -1283,11 +1401,13 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin
uploadDuration := time.Since(startTime)
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
if err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID); err != nil {
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
if err != nil {
return fmt.Errorf("updating blob upload timestamp: %w", err)
}
if err := s.repos.Snapshots.AddBlob(txCtx, tx, s.snapshotID, finishedBlobID, types.BlobHash(finishedBlob.Hash)); err != nil {
err = s.repos.Snapshots.AddBlob(txCtx, tx, s.snapshotID, finishedBlobID, types.BlobHash(finishedBlob.Hash))
if err != nil {
return fmt.Errorf("adding blob to snapshot: %w", err)
}
@@ -1299,7 +1419,9 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin
Size: finishedBlob.Compressed,
DurationMs: uploadDuration.Milliseconds(),
}
if err := s.repos.Uploads.Create(txCtx, tx, upload); err != nil {
err := s.repos.Uploads.Create(txCtx, tx, upload)
if err != nil {
return fmt.Errorf("recording upload metrics: %w", err)
}
}
@@ -1312,10 +1434,14 @@ func (s *Scanner) recordBlobMetadata(ctx context.Context, finishedBlob *blob.Fin
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.BlobWithReader) {
if blobWithReader.TempFile != nil {
tempName := blobWithReader.TempFile.Name()
if err := blobWithReader.TempFile.Close(); err != nil {
err := blobWithReader.TempFile.Close()
if err != nil {
log.Fatal("Failed to close temp file", "file", tempName, "error", err)
}
if err := s.fs.Remove(tempName); err != nil {
err = s.fs.Remove(tempName)
if err != nil {
log.Fatal("Failed to remove temp file", "file", tempName, "error", err)
}
}
@@ -1343,6 +1469,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT
defer func() { _ = file.Close() }()
var chunks []streamingChunkInfo
chunkIndex := 0
fileHash, err := s.chunker.ChunkReaderStreaming(file, func(chunk chunker.Chunk) error {
@@ -1372,16 +1499,17 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT
s.updateChunkStats(chunkExists, chunk.Size, result)
if !chunkExists {
if err := s.addChunkToPacker(chunk); err != nil {
err := s.addChunkToPacker(chunk)
if err != nil {
return err
}
}
chunk.Data = nil
chunkIndex++
return nil
})
if err != nil {
return fmt.Errorf("chunking file: %w", err)
}
@@ -1390,6 +1518,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT
"path", fileToProcess.Path, "file_hash", fileHash, "chunks", len(chunks))
s.queueFileForBatchInsert(ctx, fileToProcess, chunks)
return nil
}
@@ -1397,6 +1526,7 @@ func (s *Scanner) processFileStreaming(ctx context.Context, fileToProcess *FileT
func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *ScanResult) {
if chunkExists {
result.FilesSkipped++
result.BytesSkipped += chunkSize
if s.progress != nil {
s.progress.GetStats().BytesSkipped.Add(chunkSize)
@@ -1404,6 +1534,7 @@ func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *Sc
} else {
result.ChunksCreated++
result.BytesScanned += chunkSize
if s.progress != nil {
s.progress.GetStats().ChunksCreated.Add(1)
s.progress.GetStats().BytesProcessed.Add(chunkSize)
@@ -1415,27 +1546,36 @@ func (s *Scanner) updateChunkStats(chunkExists bool, chunkSize int64, result *Sc
// addChunkToPacker adds a chunk to the blob packer, finalizing the current blob if needed
func (s *Scanner) addChunkToPacker(chunk chunker.Chunk) error {
s.packerMu.Lock()
err := s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data})
if err == blob.ErrBlobSizeLimitExceeded {
if err := s.packer.FinalizeBlob(); err != nil {
if errors.Is(err, blob.ErrBlobSizeLimitExceeded) {
err := s.packer.FinalizeBlob()
if err != nil {
s.packerMu.Unlock()
return fmt.Errorf("finalizing blob: %w", err)
}
if err := s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data}); err != nil {
err = s.packer.AddChunk(&blob.ChunkRef{Hash: chunk.Hash, Data: chunk.Data})
if err != nil {
s.packerMu.Unlock()
return fmt.Errorf("adding chunk after finalize: %w", err)
}
} else if err != nil {
s.packerMu.Unlock()
return fmt.Errorf("adding chunk to packer: %w", err)
}
s.packerMu.Unlock()
return nil
}
// queueFileForBatchInsert builds file/chunk associations and queues the file for batch DB insert
func (s *Scanner) queueFileForBatchInsert(ctx context.Context, fileToProcess *FileToProcess, chunks []streamingChunkInfo) {
fileChunks := make([]database.FileChunk, len(chunks))
chunkFiles := make([]database.ChunkFile, len(chunks))
for i, ci := range chunks {
fileChunks[i] = database.FileChunk{
@@ -1503,6 +1643,7 @@ func wrapPermissionError(path string, err error) error {
if !errors.Is(err, os.ErrPermission) {
return err
}
if runtime.GOOS == "darwin" {
return fmt.Errorf("cannot read %s: %w\n\n"+
"macOS is blocking access to this path. Grant Full Disk Access to your\n"+
@@ -1510,12 +1651,14 @@ func wrapPermissionError(path string, err error) error {
" System Settings → Privacy & Security → Full Disk Access\n\n"+
"then quit and reopen the terminal and re-run the backup", path, err)
}
return fmt.Errorf("cannot read %s: %w (check file permissions, or run with --skip-errors to continue past unreadable files)", path, err)
}
// compileExcludePatterns compiles the exclude patterns into glob matchers
func compileExcludePatterns(patterns []string) []compiledPattern {
var compiled []compiledPattern
for _, p := range patterns {
if p == "" {
continue
@@ -1523,6 +1666,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern {
// Check if pattern is anchored (starts with /)
anchored := strings.HasPrefix(p, "/")
pattern := p
if anchored {
pattern = p[1:] // Remove leading /
@@ -1537,6 +1681,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern {
g, err := glob.Compile(pattern, '/')
if err != nil {
log.Warn("Invalid exclude pattern, skipping", "pattern", p, "error", err)
continue
}
@@ -1546,6 +1691,7 @@ func compileExcludePatterns(patterns []string) []compiledPattern {
original: p,
})
}
return compiled
}