Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
803be51073 |
+4
-4
@@ -279,8 +279,8 @@ Key methods:
|
|||||||
- `Scan(ctx, path, snapshotID)` → Main scan operation
|
- `Scan(ctx, path, snapshotID)` → Main scan operation
|
||||||
|
|
||||||
Scan phases:
|
Scan phases:
|
||||||
1. **Phase 0**: Detect deleted files from previous snapshots
|
1. **Phase 1**: Walk directory, identify files needing processing
|
||||||
2. **Phase 1**: Walk directory, identify files needing processing
|
2. **Phase 1b**: Count as deleted the files that the previous snapshot of the same name lists and the walk did not find; a file that is excluded, or inside an excluded directory, is not counted
|
||||||
3. **Phase 2**: Process files (chunk → pack → upload)
|
3. **Phase 2**: Process files (chunk → pack → upload)
|
||||||
|
|
||||||
#### SnapshotManager
|
#### SnapshotManager
|
||||||
@@ -320,13 +320,13 @@ CreateSnapshot(opts)
|
|||||||
│ │
|
│ │
|
||||||
│ ├─► scanner.Scan(ctx, path, snapshotID)
|
│ ├─► scanner.Scan(ctx, path, snapshotID)
|
||||||
│ │ │
|
│ │ │
|
||||||
│ │ ├─► Phase 0: detectDeletedFiles()
|
|
||||||
│ │ │
|
|
||||||
│ │ ├─► Phase 1: scanPhase()
|
│ │ ├─► Phase 1: scanPhase()
|
||||||
│ │ │ Walk directory
|
│ │ │ Walk directory
|
||||||
│ │ │ Check file metadata changes
|
│ │ │ Check file metadata changes
|
||||||
│ │ │ Build list of files to process
|
│ │ │ Build list of files to process
|
||||||
│ │ │
|
│ │ │
|
||||||
|
│ │ ├─► Phase 1b: detectDeletedFiles()
|
||||||
|
│ │ │
|
||||||
│ │ └─► Phase 2: processPhase()
|
│ │ └─► Phase 2: processPhase()
|
||||||
│ │ For each file:
|
│ │ For each file:
|
||||||
│ │ chunker.ChunkReaderStreaming()
|
│ │ chunker.ChunkReaderStreaming()
|
||||||
|
|||||||
@@ -267,6 +267,45 @@ func (r *FileRepository) ListUnderPath(
|
|||||||
return files, rows.Err()
|
return files, rows.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ListUnderPathInSnapshot returns the files that ListUnderPath returns
|
||||||
|
// for path and that the snapshot lists, in no particular order.
|
||||||
|
func (r *FileRepository) ListUnderPathInSnapshot(
|
||||||
|
ctx context.Context, path, snapshotID string,
|
||||||
|
) ([]*File, error) {
|
||||||
|
path = strings.TrimRight(path, "/")
|
||||||
|
dirPrefix := path + "/"
|
||||||
|
|
||||||
|
query := `
|
||||||
|
SELECT f.id, f.path, f.source_path, f.mtime, f.mtime_nsec, f.size,
|
||||||
|
f.mode, f.uid, f.gid, f.link_target
|
||||||
|
FROM files f
|
||||||
|
JOIN snapshot_files sf ON sf.file_id = f.id
|
||||||
|
WHERE sf.snapshot_id = ?
|
||||||
|
AND (f.path = ? OR substr(f.path, 1, length(?)) = ?)
|
||||||
|
`
|
||||||
|
|
||||||
|
rows, err := r.db.conn.QueryContext(ctx, query,
|
||||||
|
snapshotID, path, dirPrefix, dirPrefix)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("querying files: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
defer func() { _ = rows.Close() }()
|
||||||
|
|
||||||
|
var files []*File
|
||||||
|
|
||||||
|
for rows.Next() {
|
||||||
|
file, err := r.scanFileRows(rows)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("scanning file: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
files = append(files, file)
|
||||||
|
}
|
||||||
|
|
||||||
|
return files, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
// ListIDsWithChunksNotInUploadedBlobs returns the IDs of the files whose
|
// ListIDsWithChunksNotInUploadedBlobs returns the IDs of the files whose
|
||||||
// path starts with prefix and that list at least one chunk held by no
|
// path starts with prefix and that list at least one chunk held by no
|
||||||
// blob whose upload has completed (uploaded_ts set). A new snapshot
|
// blob whose upload has completed (uploaded_ts set). A new snapshot
|
||||||
|
|||||||
@@ -212,6 +212,65 @@ func TestFileRepositoryListUnderPath(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ListUnderPathInSnapshot leaves out files outside path and files the
|
||||||
|
// snapshot does not list, though another snapshot lists them.
|
||||||
|
func TestFileRepositoryListUnderPathInSnapshot(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
files := database.NewFileRepository(db)
|
||||||
|
snapshots := database.NewSnapshotRepository(db)
|
||||||
|
|
||||||
|
listed := map[string][]string{
|
||||||
|
"one": {"/src/a.txt", "/src2/c.txt"},
|
||||||
|
"two": {"/src/b.txt"},
|
||||||
|
}
|
||||||
|
|
||||||
|
for snapshotID, paths := range listed {
|
||||||
|
err := snapshots.Create(ctx, nil, &database.Snapshot{
|
||||||
|
ID: types.SnapshotID(snapshotID),
|
||||||
|
Hostname: "host",
|
||||||
|
StartedAt: time.Now(),
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create snapshot %s: %v", snapshotID, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, path := range paths {
|
||||||
|
err := files.Create(ctx, nil, &database.File{
|
||||||
|
Path: types.FilePath(path),
|
||||||
|
MTime: time.Now().Truncate(time.Second),
|
||||||
|
Mode: 0644,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create %s: %v", path, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = snapshots.AddFile(ctx, nil, snapshotID, path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to add %s to %s: %v", path, snapshotID, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
found, err := files.ListUnderPathInSnapshot(ctx, "/src", "one")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to list files: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
got := make([]string, 0, len(found))
|
||||||
|
for _, f := range found {
|
||||||
|
got = append(got, f.Path.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
if !slices.Equal(got, []string{"/src/a.txt"}) {
|
||||||
|
t.Errorf("got %q, want only /src/a.txt", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestFileRepositorySymlink(t *testing.T) {
|
func TestFileRepositorySymlink(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
+15
-11
@@ -16,6 +16,9 @@ type ScannerParams struct {
|
|||||||
Fs afero.Fs
|
Fs afero.Fs
|
||||||
Exclude []string // Exclude patterns (combined global + snapshot-specific)
|
Exclude []string // Exclude patterns (combined global + snapshot-specific)
|
||||||
SkipErrors bool // Skip file read errors (log loudly but continue)
|
SkipErrors bool // Skip file read errors (log loudly but continue)
|
||||||
|
// PreviousSnapshotID is the most recent completed snapshot of the
|
||||||
|
// same name, "" for none.
|
||||||
|
PreviousSnapshotID string
|
||||||
}
|
}
|
||||||
|
|
||||||
// Module exports backup functionality as an fx module.
|
// Module exports backup functionality as an fx module.
|
||||||
@@ -44,17 +47,18 @@ func provideScannerFactory(
|
|||||||
}
|
}
|
||||||
|
|
||||||
return NewScanner(ScannerConfig{
|
return NewScanner(ScannerConfig{
|
||||||
FS: params.Fs,
|
FS: params.Fs,
|
||||||
ChunkSize: cfg.ChunkSize.Int64(),
|
ChunkSize: cfg.ChunkSize.Int64(),
|
||||||
Repositories: repos,
|
Repositories: repos,
|
||||||
Storage: storer,
|
Storage: storer,
|
||||||
MaxBlobSize: cfg.BlobSizeLimit.Int64(),
|
MaxBlobSize: cfg.BlobSizeLimit.Int64(),
|
||||||
CompressionLevel: cfg.CompressionLevel,
|
CompressionLevel: cfg.CompressionLevel,
|
||||||
AgeRecipients: cfg.AgeRecipients,
|
AgeRecipients: cfg.AgeRecipients,
|
||||||
EnableProgress: params.EnableProgress,
|
EnableProgress: params.EnableProgress,
|
||||||
UI: params.UI,
|
UI: params.UI,
|
||||||
Exclude: excludes,
|
Exclude: excludes,
|
||||||
SkipErrors: params.SkipErrors,
|
SkipErrors: params.SkipErrors,
|
||||||
|
PreviousSnapshotID: params.PreviousSnapshotID,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -58,6 +58,10 @@ type Scanner struct {
|
|||||||
compressionLevel int
|
compressionLevel int
|
||||||
ageRecipient string
|
ageRecipient string
|
||||||
snapshotID string // Current snapshot being processed
|
snapshotID string // Current snapshot being processed
|
||||||
|
// previousSnapshotID is the most recent completed snapshot of the
|
||||||
|
// same name, "" for none. Only a file it lists can be counted as
|
||||||
|
// deleted.
|
||||||
|
previousSnapshotID string
|
||||||
// currentSourcePath is the source directory being scanned, stored with
|
// currentSourcePath is the source directory being scanned, stored with
|
||||||
// each file record.
|
// each file record.
|
||||||
currentSourcePath string
|
currentSourcePath string
|
||||||
@@ -129,6 +133,9 @@ type ScannerConfig struct {
|
|||||||
// SkipErrors skips files that cannot be opened or read (log loudly but
|
// SkipErrors skips files that cannot be opened or read (log loudly but
|
||||||
// continue); packer, database, encryption, and upload errors still abort.
|
// continue); packer, database, encryption, and upload errors still abort.
|
||||||
SkipErrors bool
|
SkipErrors bool
|
||||||
|
// PreviousSnapshotID is the most recent completed snapshot of the
|
||||||
|
// same name, "" for none.
|
||||||
|
PreviousSnapshotID string
|
||||||
}
|
}
|
||||||
|
|
||||||
// ScanResult contains the results of a scan operation. Files and bytes
|
// ScanResult contains the results of a scan operation. Files and bytes
|
||||||
@@ -207,6 +214,7 @@ func NewScanner(cfg ScannerConfig) *Scanner {
|
|||||||
compiledExclude: compiledExclude,
|
compiledExclude: compiledExclude,
|
||||||
progress: progress,
|
progress: progress,
|
||||||
skipErrors: cfg.SkipErrors,
|
skipErrors: cfg.SkipErrors,
|
||||||
|
previousSnapshotID: cfg.PreviousSnapshotID,
|
||||||
ui: uiw,
|
ui: uiw,
|
||||||
pendingChunkHashes: make(map[string]struct{}),
|
pendingChunkHashes: make(map[string]struct{}),
|
||||||
}
|
}
|
||||||
@@ -258,8 +266,9 @@ func (s *Scanner) Scan(
|
|||||||
|
|
||||||
filesToProcess := scanResult.FilesToProcess
|
filesToProcess := scanResult.FilesToProcess
|
||||||
|
|
||||||
// Phase 1b: Detect deleted files by comparing DB against scanned files
|
// Phase 1b: Detect deleted files by comparing the previous snapshot
|
||||||
err = s.detectDeletedFilesFromMap(ctx, knownFiles, existingFiles, result)
|
// of this name against scanned files
|
||||||
|
err = s.detectDeletedFiles(ctx, path, existingFiles, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("detecting deleted files: %w", err)
|
return nil, fmt.Errorf("detecting deleted files: %w", err)
|
||||||
}
|
}
|
||||||
@@ -439,9 +448,8 @@ func (s *Scanner) summarizeScanPhase(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// loadKnownFiles loads the known files at and beneath path from the
|
// loadKnownFiles loads the known files at and beneath path from the
|
||||||
// database into a map for fast lookup. Every loaded file the scan does
|
// database into a map for fast lookup. This avoids per-file database
|
||||||
// not find is counted as deleted. This avoids per-file database queries
|
// queries during the scan phase.
|
||||||
// during the scan phase.
|
|
||||||
func (s *Scanner) loadKnownFiles(
|
func (s *Scanner) loadKnownFiles(
|
||||||
ctx context.Context, path string,
|
ctx context.Context, path string,
|
||||||
) (map[string]*database.File, error) {
|
) (map[string]*database.File, error) {
|
||||||
@@ -1954,21 +1962,28 @@ func (s *Scanner) queueFileForBatchInsert(
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// detectDeletedFilesFromMap finds files that existed in previous snapshots
|
// detectDeletedFiles counts as deleted each file at or beneath path that
|
||||||
// but no longer exist. Uses pre-loaded maps to avoid any filesystem or
|
// the previous snapshot of this name lists and the walk did not find.
|
||||||
// database access.
|
// A file that is excluded, or inside an excluded directory, is not
|
||||||
func (s *Scanner) detectDeletedFilesFromMap(
|
// counted: the walk skips it whether or not it is still on disk.
|
||||||
|
func (s *Scanner) detectDeletedFiles(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
knownFiles map[string]*database.File,
|
path string,
|
||||||
existingFiles map[string]struct{},
|
existingFiles map[string]struct{},
|
||||||
result *ScanResult,
|
result *ScanResult,
|
||||||
) error {
|
) error {
|
||||||
if len(knownFiles) == 0 {
|
if s.previousSnapshotID == "" {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check each known file against the enumerated set (no filesystem access needed)
|
previousFiles, err := s.repos.Files.ListUnderPathInSnapshot(
|
||||||
for path, file := range knownFiles {
|
ctx, path, s.previousSnapshotID)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("listing files of snapshot %s: %w",
|
||||||
|
s.previousSnapshotID, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, file := range previousFiles {
|
||||||
// Check context cancellation periodically
|
// Check context cancellation periodically
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
@@ -1976,13 +1991,16 @@ func (s *Scanner) detectDeletedFilesFromMap(
|
|||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check if the file exists in our enumerated set
|
filePath := file.Path.String()
|
||||||
if _, exists := existingFiles[path]; !exists {
|
|
||||||
// File has been deleted
|
_, exists := existingFiles[filePath]
|
||||||
result.FilesDeleted++
|
if exists || s.excludedOrInExcludedDir(filePath, path) {
|
||||||
result.BytesDeleted += file.Size
|
continue
|
||||||
log.Debug("Detected deleted file", "path", path, "size", file.Size)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
result.FilesDeleted++
|
||||||
|
result.BytesDeleted += file.Size
|
||||||
|
log.Debug("Detected deleted file", "path", filePath, "size", file.Size)
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.FilesDeleted > 0 {
|
if result.FilesDeleted > 0 {
|
||||||
@@ -1993,6 +2011,20 @@ func (s *Scanner) detectDeletedFilesFromMap(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// excludedOrInExcludedDir reports whether filePath, which is at or
|
||||||
|
// beneath rootPath, or any directory between rootPath and it, matches an
|
||||||
|
// exclude pattern. A pattern such as */.local/share/Trash matches the
|
||||||
|
// directory but not the files inside it, which the walk skips too.
|
||||||
|
func (s *Scanner) excludedOrInExcludedDir(filePath, rootPath string) bool {
|
||||||
|
for p := filePath; len(p) > len(rootPath); p = filepath.Dir(p) {
|
||||||
|
if s.shouldExclude(p, rootPath) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// wrapPermissionError augments permission errors with platform-specific
|
// wrapPermissionError augments permission errors with platform-specific
|
||||||
// remediation instructions. On macOS, TCC-protected directories (Calendars,
|
// remediation instructions. On macOS, TCC-protected directories (Calendars,
|
||||||
// Reminders, Photos, etc.) return EPERM unless the running application has
|
// Reminders, Photos, etc.) return EPERM unless the running application has
|
||||||
|
|||||||
+13
-63
@@ -120,10 +120,9 @@ func (f *FileStorer) Delete(_ context.Context, key string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// List returns all keys with the given prefix, reading through symlinked
|
// List returns all keys with the given prefix. It fails when the
|
||||||
// directories. It fails when the destination directory is missing, or
|
// destination directory is missing; a missing prefix under it is an
|
||||||
// when the prefix or a directory above it is a symlink whose target is
|
// empty listing.
|
||||||
// missing; a prefix that is truly missing is an empty listing.
|
|
||||||
func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
|
func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
|
||||||
var keys []string
|
var keys []string
|
||||||
|
|
||||||
@@ -134,7 +133,8 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
exists, err := f.prefixExists(prefix)
|
// Check if the prefix exists
|
||||||
|
exists, err := afero.Exists(f.fs, basePath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("checking path: %w", err)
|
return nil, fmt.Errorf("checking path: %w", err)
|
||||||
}
|
}
|
||||||
@@ -143,7 +143,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
return keys, nil // Empty list for non-existent prefix
|
return keys, nil // Empty list for non-existent prefix
|
||||||
}
|
}
|
||||||
|
|
||||||
err = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -176,9 +176,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ListStream returns a channel of ObjectInfo for large result sets. Like
|
// ListStream returns a channel of ObjectInfo for large result sets. Like
|
||||||
// List, it reads through symlinked directories and sends an error when
|
// List, it sends an error when the destination directory is missing.
|
||||||
// the destination directory is missing, or when the prefix or a directory
|
|
||||||
// above it is a symlink whose target is missing.
|
|
||||||
func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan ObjectInfo {
|
func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan ObjectInfo {
|
||||||
ch := make(chan ObjectInfo)
|
ch := make(chan ObjectInfo)
|
||||||
|
|
||||||
@@ -194,7 +192,8 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
exists, err := f.prefixExists(prefix)
|
// Check if the prefix exists
|
||||||
|
exists, err := afero.Exists(f.fs, basePath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
ch <- ObjectInfo{Err: fmt.Errorf("checking path: %w", err)}
|
ch <- ObjectInfo{Err: fmt.Errorf("checking path: %w", err)}
|
||||||
|
|
||||||
@@ -205,7 +204,7 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
return // Empty channel for non-existent prefix
|
return // Empty channel for non-existent prefix
|
||||||
}
|
}
|
||||||
|
|
||||||
_ = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
_ = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
// Check context cancellation
|
// Check context cancellation
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
@@ -244,12 +243,11 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
}
|
}
|
||||||
|
|
||||||
// DeletePartialUploads removes every file under prefix whose name ends in
|
// DeletePartialUploads removes every file under prefix whose name ends in
|
||||||
// tempSuffix. Like List, it reads through symlinked directories and fails
|
// tempSuffix. A missing prefix has none to remove.
|
||||||
// on one whose target is missing. A missing prefix has none to remove.
|
|
||||||
func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
exists, err := f.prefixExists(prefix)
|
exists, err := afero.Exists(f.fs, basePath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("checking path: %w", err)
|
return fmt.Errorf("checking path: %w", err)
|
||||||
}
|
}
|
||||||
@@ -258,7 +256,7 @@ func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) er
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
err = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -354,54 +352,6 @@ func (f *FileStorer) writeAtomic(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// walk is afero.Walk, except that it reads through a symlinked directory
|
|
||||||
// the way Get and Stat do, where afero.Walk reports the link as one file.
|
|
||||||
// afero.Walk uses Lstat, which does not follow a link, whenever the
|
|
||||||
// filesystem has one; the struct below carries only the afero.Fs methods,
|
|
||||||
// so the walk uses Stat instead. A link back up the tree is followed until
|
|
||||||
// the operating system refuses the path for too many levels of links, and
|
|
||||||
// the walk then fails.
|
|
||||||
func (f *FileStorer) walk(root string, fn filepath.WalkFunc) error {
|
|
||||||
return afero.Walk(struct{ afero.Fs }{f.fs}, root, fn)
|
|
||||||
}
|
|
||||||
|
|
||||||
// prefixExists reports whether prefix exists under the destination
|
|
||||||
// directory. It checks each directory on the way down to prefix in turn,
|
|
||||||
// and the first one Stat cannot find decides: if it is not there at all,
|
|
||||||
// prefix is missing; if it is a symlink whose target is missing, such as
|
|
||||||
// a link to a volume that is not mounted, prefixExists fails, since
|
|
||||||
// listing it as empty would let prune delete every blob.
|
|
||||||
func (f *FileStorer) prefixExists(prefix string) (bool, error) {
|
|
||||||
lstater, canLstat := f.fs.(afero.Lstater)
|
|
||||||
path := f.basePath
|
|
||||||
|
|
||||||
for name := range strings.SplitSeq(prefix, "/") {
|
|
||||||
path = filepath.Join(path, name)
|
|
||||||
|
|
||||||
_, err := f.fs.Stat(path)
|
|
||||||
if err == nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
if !os.IsNotExist(err) {
|
|
||||||
return false, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Lstat does not follow a symlink at path, so it finds the link
|
|
||||||
// that Stat could not follow.
|
|
||||||
if canLstat {
|
|
||||||
_, _, lstatErr := lstater.LstatIfPossible(path)
|
|
||||||
if lstatErr == nil {
|
|
||||||
return false, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return false, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
return true, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// fullPath returns the full filesystem path for a key.
|
// fullPath returns the full filesystem path for a key.
|
||||||
func (f *FileStorer) fullPath(key string) string {
|
func (f *FileStorer) fullPath(key string) string {
|
||||||
return filepath.Join(f.basePath, key)
|
return filepath.Join(f.basePath, key)
|
||||||
|
|||||||
@@ -4,10 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"reflect"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
@@ -63,232 +60,3 @@ func TestFileStorerListMissingDestination(t *testing.T) {
|
|||||||
t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
|
t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// The two directories at the top of a destination, either of which the
|
|
||||||
// tests below replace with a symlink.
|
|
||||||
const (
|
|
||||||
metadataDir = "metadata"
|
|
||||||
blobsDir = "blobs"
|
|
||||||
)
|
|
||||||
|
|
||||||
// moveBehindSymlink moves base/dir to a directory elsewhere and puts a
|
|
||||||
// symlink to it in its place.
|
|
||||||
func moveBehindSymlink(t *testing.T, base, dir string) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
linkPath := filepath.Join(base, dir)
|
|
||||||
movedPath := filepath.Join(t.TempDir(), dir)
|
|
||||||
|
|
||||||
err := os.Rename(linkPath, movedPath)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("moving %s away: %v", dir, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = os.Symlink(movedPath, linkPath)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("linking %s: %v", dir, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestFileStorerListsThroughSymlinkedDirectory checks that List and
|
|
||||||
// ListStream read through a symlinked metadata/ or blobs/ directory, as Get
|
|
||||||
// and Stat do, instead of listing the link as one key. Listing the whole
|
|
||||||
// destination meets the link part-way through the walk; listing its prefix
|
|
||||||
// starts the walk at the link. A prefix missing behind the link lists as
|
|
||||||
// empty.
|
|
||||||
func TestFileStorerListsThroughSymlinkedDirectory(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
for _, dir := range []string{metadataDir, blobsDir} {
|
|
||||||
t.Run(dir, func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
base := t.TempDir()
|
|
||||||
key := dir + "/aa/bb/object"
|
|
||||||
|
|
||||||
s, err := storage.NewFileStorer(base)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("NewFileStorer: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = s.Put(context.Background(), key, strings.NewReader("data"))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Put: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
moveBehindSymlink(t, base, dir)
|
|
||||||
|
|
||||||
want := []string{key}
|
|
||||||
|
|
||||||
for _, prefix := range []string{"", dir + "/"} {
|
|
||||||
got := listSorted(t, s, prefix)
|
|
||||||
if !reflect.DeepEqual(got, want) {
|
|
||||||
t.Errorf("List(%q) = %v, want %v", prefix, got, want)
|
|
||||||
}
|
|
||||||
|
|
||||||
got = listStreamKeys(t, s, prefix)
|
|
||||||
if !reflect.DeepEqual(got, want) {
|
|
||||||
t.Errorf("ListStream(%q) = %v, want %v", prefix, got, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
absent := dir + "/absent/"
|
|
||||||
|
|
||||||
if got := listSorted(t, s, absent); len(got) != 0 {
|
|
||||||
t.Errorf("List(%q) = %v, want empty", absent, got)
|
|
||||||
}
|
|
||||||
|
|
||||||
if got := listStreamKeys(t, s, absent); len(got) != 0 {
|
|
||||||
t.Errorf("ListStream(%q) = %v, want empty", absent, got)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestFileStorerListFailsOnSymlinkWithMissingTarget checks that List and
|
|
||||||
// ListStream fail on a symlinked metadata/ or blobs/ directory whose target
|
|
||||||
// is missing, such as a link to a volume that is not mounted, as they do on
|
|
||||||
// a missing destination directory, instead of listing it as empty. That
|
|
||||||
// holds for a prefix below the link too.
|
|
||||||
func TestFileStorerListFailsOnSymlinkWithMissingTarget(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
for _, dir := range []string{metadataDir, blobsDir} {
|
|
||||||
t.Run(dir, func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
base := t.TempDir()
|
|
||||||
|
|
||||||
s, err := storage.NewFileStorer(base)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("NewFileStorer: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
unmounted := filepath.Join(t.TempDir(), "unmounted")
|
|
||||||
|
|
||||||
err = os.Symlink(unmounted, filepath.Join(base, dir))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Symlink: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, prefix := range []string{"", dir + "/", dir + "/aa/"} {
|
|
||||||
keys, err := s.List(ctx, prefix)
|
|
||||||
if !errors.Is(err, fs.ErrNotExist) {
|
|
||||||
t.Errorf("List(%q) = %v, %v; want a not-exist error",
|
|
||||||
prefix, keys, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
var streamErr error
|
|
||||||
|
|
||||||
for object := range s.ListStream(ctx, prefix) {
|
|
||||||
if object.Err != nil {
|
|
||||||
streamErr = object.Err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if !errors.Is(streamErr, fs.ErrNotExist) {
|
|
||||||
t.Errorf("ListStream(%q) error = %v, want a not-exist error",
|
|
||||||
prefix, streamErr)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestFileStorerListFailsOnSymlinkLoop checks that a symlink pointing back
|
|
||||||
// at the destination makes List and ListStream fail instead of walking it
|
|
||||||
// forever.
|
|
||||||
func TestFileStorerListFailsOnSymlinkLoop(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
base := t.TempDir()
|
|
||||||
|
|
||||||
s, err := storage.NewFileStorer(base)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("NewFileStorer: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = s.Put(ctx, testBlobKey, strings.NewReader("data"))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Put: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = os.Symlink(base, filepath.Join(base, "blobs", "loop"))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Symlink: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
keys, err := s.List(ctx, "blobs/")
|
|
||||||
if err == nil {
|
|
||||||
t.Errorf("List = %d keys and no error, want an error", len(keys))
|
|
||||||
}
|
|
||||||
|
|
||||||
var streamErr error
|
|
||||||
|
|
||||||
for object := range s.ListStream(ctx, "blobs/") {
|
|
||||||
if object.Err != nil {
|
|
||||||
streamErr = object.Err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if streamErr == nil {
|
|
||||||
t.Error("ListStream sent no error, want one")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory checks that
|
|
||||||
// DeletePartialUploads, which remote nuke uses, removes a leftover temp
|
|
||||||
// file behind a symlinked metadata/ or blobs/ directory and keeps the
|
|
||||||
// object at the real key.
|
|
||||||
func TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
for _, dir := range []string{metadataDir, blobsDir} {
|
|
||||||
t.Run(dir, func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
base := t.TempDir()
|
|
||||||
key := dir + "/aa/bb/object"
|
|
||||||
|
|
||||||
s, err := storage.NewFileStorer(base)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("NewFileStorer: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = s.Put(ctx, key, strings.NewReader("data"))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("Put: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
leftover := filepath.Join(base, key+"-123456.partial")
|
|
||||||
|
|
||||||
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("writing leftover temp file: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
moveBehindSymlink(t, base, dir)
|
|
||||||
|
|
||||||
err = s.DeletePartialUploads(ctx, dir+"/")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("DeletePartialUploads: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// leftover runs through the link.
|
|
||||||
_, err = os.Stat(leftover)
|
|
||||||
if !os.IsNotExist(err) {
|
|
||||||
t.Errorf("leftover temp file was not removed: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = s.Stat(ctx, key)
|
|
||||||
if err != nil {
|
|
||||||
t.Errorf("Stat of the real key: %v", err)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,9 +1,7 @@
|
|||||||
package vaultik_test
|
package vaultik_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"crypto/rand"
|
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -11,9 +9,7 @@ import (
|
|||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"sneak.berlin/go/vaultik/internal/config"
|
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
@@ -81,174 +77,6 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
|||||||
"deep verify should fail on a corrupted blob")
|
"deep verify should fail on a corrupted blob")
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
|
|
||||||
// that deep verify reports under --json: on a healthy snapshot, with a
|
|
||||||
// damaged blob, and with missing blobs. Deep verify stops at the first
|
|
||||||
// damaged blob, so the blobs read before it are the ones verified; when
|
|
||||||
// blobs are missing it reads no blob contents, but counts every missing
|
|
||||||
// blob.
|
|
||||||
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
fs := afero.NewOsFs()
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
|
||||||
|
|
||||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
|
||||||
|
|
||||||
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Equal(t, "ok", healthy.Status)
|
|
||||||
require.Len(t, keys, healthy.BlobCount)
|
|
||||||
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
|
||||||
require.Zero(t, healthy.Missing)
|
|
||||||
|
|
||||||
// Damage the last blob read, keeping its length, so every blob before
|
|
||||||
// it is read and passes first.
|
|
||||||
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
|
|
||||||
blob, err := afero.ReadFile(fs, lastPath)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
blob[len(blob)/2] ^= 0xff
|
|
||||||
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
|
|
||||||
|
|
||||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
|
||||||
require.Error(t, err)
|
|
||||||
require.Equal(t, "failed", got.Status)
|
|
||||||
require.Equal(t, healthy.BlobCount-1, got.Verified)
|
|
||||||
require.Zero(t, got.Missing)
|
|
||||||
|
|
||||||
// Delete the first blob and the damaged last one: both must be
|
|
||||||
// counted as missing, not only the first one found.
|
|
||||||
var deletedSize int64
|
|
||||||
|
|
||||||
for _, key := range []string{keys[0], keys[len(keys)-1]} {
|
|
||||||
info, statErr := storer.Stat(ctx, key)
|
|
||||||
require.NoError(t, statErr)
|
|
||||||
|
|
||||||
deletedSize += info.Size
|
|
||||||
|
|
||||||
require.NoError(t, storer.Delete(ctx, key))
|
|
||||||
}
|
|
||||||
|
|
||||||
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
|
||||||
require.Error(t, err)
|
|
||||||
require.Equal(t, "failed", got.Status)
|
|
||||||
require.Equal(t, 2, got.Missing)
|
|
||||||
require.Equal(t, deletedSize, got.MissingSize)
|
|
||||||
require.Zero(t, got.Verified)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestDeepVerifyJSONCountsWrongSizeBlob checks that under --deep --json a
|
|
||||||
// blob of the wrong size is counted in mismatched and fails the run, and
|
|
||||||
// that the check goes on past it to count a missing blob after it.
|
|
||||||
func TestDeepVerifyJSONCountsWrongSizeBlob(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
fs := afero.NewOsFs()
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
|
||||||
|
|
||||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
|
||||||
|
|
||||||
// Make the first blob checked longer, and delete the last one.
|
|
||||||
firstPath := filepath.Join(storeDir, keys[0])
|
|
||||||
blob, err := afero.ReadFile(fs, firstPath)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
blob = append(blob, []byte("extra")...)
|
|
||||||
require.NoError(t, afero.WriteFile(fs, firstPath, blob, 0o600))
|
|
||||||
require.NoError(t, storer.Delete(ctx, keys[len(keys)-1]))
|
|
||||||
|
|
||||||
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
|
||||||
require.Error(t, err)
|
|
||||||
require.Equal(t, "failed", got.Status)
|
|
||||||
require.Equal(t, 1, got.Mismatched)
|
|
||||||
require.Equal(t, 1, got.Missing)
|
|
||||||
require.Zero(t, got.Verified)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestDeepVerifyStopsOnBlobThatCannotBeChecked checks that a blob that is
|
|
||||||
// present in storage but cannot be checked is not counted as missing:
|
|
||||||
// deep verify stops and fails with the storage error.
|
|
||||||
func TestDeepVerifyStopsOnBlobThatCannotBeChecked(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "remote")
|
|
||||||
|
|
||||||
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
|
||||||
failing := statFailingStorer{Storer: storer, failKeys: []string{keys[0]}}
|
|
||||||
|
|
||||||
got, err := deepVerifyJSON(ctx, t, cfg, failing, snapshotID)
|
|
||||||
require.Error(t, err)
|
|
||||||
require.Equal(t, "failed", got.Status)
|
|
||||||
require.Zero(t, got.Missing)
|
|
||||||
require.Contains(t, got.ErrorMessage, errStatDenied.Error())
|
|
||||||
}
|
|
||||||
|
|
||||||
// backUpSeveralBlobs backs up random data, which does not compress or
|
|
||||||
// deduplicate, so it fills at least three blobs in a store on disk at
|
|
||||||
// storeDir. It returns the blobs' storage keys sorted, which is hash
|
|
||||||
// order, the order deep verify reads them in.
|
|
||||||
func backUpSeveralBlobs(
|
|
||||||
ctx context.Context, t *testing.T, storeDir string,
|
|
||||||
) (*config.Config, *storage.FileStorer, string, []string) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
fs := afero.NewOsFs()
|
|
||||||
tempDir := t.TempDir()
|
|
||||||
dataDir := filepath.Join(tempDir, "source")
|
|
||||||
dbPath := filepath.Join(tempDir, "index.sqlite")
|
|
||||||
|
|
||||||
chunkSize := int64(32 * 1024)
|
|
||||||
maxBlobSize := int64(128 * 1024)
|
|
||||||
|
|
||||||
data := make([]byte, maxBlobSize*4)
|
|
||||||
_, err := rand.Read(data)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
|
||||||
require.NoError(t, afero.WriteFile(fs,
|
|
||||||
filepath.Join(dataDir, "data.bin"), data, 0o644))
|
|
||||||
|
|
||||||
cfg, storer, snapshotID := runFileStorageBackup(
|
|
||||||
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
|
||||||
|
|
||||||
keys := storedBlobKeys(t, storeDir)
|
|
||||||
require.GreaterOrEqual(t, len(keys), 3)
|
|
||||||
|
|
||||||
return cfg, storer, snapshotID, keys
|
|
||||||
}
|
|
||||||
|
|
||||||
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
|
|
||||||
// the decoded result and the error RunDeepVerify returned.
|
|
||||||
func deepVerifyJSON(
|
|
||||||
ctx context.Context, t *testing.T,
|
|
||||||
cfg *config.Config, storer storage.Storer, snapshotID string,
|
|
||||||
) (vaultik.VerifyResult, error) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var out bytes.Buffer
|
|
||||||
|
|
||||||
v := &vaultik.Vaultik{
|
|
||||||
Config: cfg,
|
|
||||||
Storage: storer,
|
|
||||||
Fs: afero.NewOsFs(),
|
|
||||||
Stdout: &out,
|
|
||||||
Stderr: io.Discard,
|
|
||||||
UI: ui.NewWithColor(io.Discard, false),
|
|
||||||
}
|
|
||||||
v.SetContext(ctx)
|
|
||||||
|
|
||||||
err := v.RunDeepVerify(snapshotID,
|
|
||||||
&vaultik.VerifyOptions{Deep: true, JSON: true})
|
|
||||||
|
|
||||||
return decodeVerifyResult(t, out.Bytes()), err
|
|
||||||
}
|
|
||||||
|
|
||||||
// corruptOneBlob flips a middle byte of the first blob file found under
|
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||||
// blobsDir, leaving the file length unchanged.
|
// blobsDir, leaving the file length unchanged.
|
||||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||||
|
|||||||
@@ -661,17 +661,18 @@ func faultScannerFactory(
|
|||||||
) snapshot.ScannerFactory {
|
) snapshot.ScannerFactory {
|
||||||
return func(params snapshot.ScannerParams) *snapshot.Scanner {
|
return func(params snapshot.ScannerParams) *snapshot.Scanner {
|
||||||
return snapshot.NewScanner(snapshot.ScannerConfig{
|
return snapshot.NewScanner(snapshot.ScannerConfig{
|
||||||
FS: params.Fs,
|
FS: params.Fs,
|
||||||
Storage: storer,
|
Storage: storer,
|
||||||
ChunkSize: faultChunkSize,
|
ChunkSize: faultChunkSize,
|
||||||
MaxBlobSize: faultMaxBlobSize,
|
MaxBlobSize: faultMaxBlobSize,
|
||||||
CompressionLevel: cfg.CompressionLevel,
|
CompressionLevel: cfg.CompressionLevel,
|
||||||
AgeRecipients: cfg.AgeRecipients,
|
AgeRecipients: cfg.AgeRecipients,
|
||||||
Repositories: repos,
|
Repositories: repos,
|
||||||
EnableProgress: params.EnableProgress,
|
EnableProgress: params.EnableProgress,
|
||||||
UI: params.UI,
|
UI: params.UI,
|
||||||
Exclude: params.Exclude,
|
Exclude: params.Exclude,
|
||||||
SkipErrors: params.SkipErrors,
|
SkipErrors: params.SkipErrors,
|
||||||
|
PreviousSnapshotID: params.PreviousSnapshotID,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -243,12 +243,18 @@ func (v *Vaultik) createNamedSnapshot(
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
previousSnapshotID, err := v.previousSnapshotID(snapName)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
scanner := v.ScannerFactory(snapshot.ScannerParams{
|
scanner := v.ScannerFactory(snapshot.ScannerParams{
|
||||||
EnableProgress: !opts.Cron,
|
EnableProgress: !opts.Cron,
|
||||||
UI: v.UI,
|
UI: v.UI,
|
||||||
Fs: v.Fs,
|
Fs: v.Fs,
|
||||||
Exclude: v.Config.GetExcludes(snapName),
|
Exclude: v.Config.GetExcludes(snapName),
|
||||||
SkipErrors: opts.SkipErrors,
|
SkipErrors: opts.SkipErrors,
|
||||||
|
PreviousSnapshotID: previousSnapshotID,
|
||||||
})
|
})
|
||||||
|
|
||||||
snapshotID, err := v.SnapshotManager.CreateSnapshotWithName(
|
snapshotID, err := v.SnapshotManager.CreateSnapshotWithName(
|
||||||
@@ -283,6 +289,24 @@ func (v *Vaultik) createNamedSnapshot(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// previousSnapshotID returns the ID of the most recent completed snapshot
|
||||||
|
// named snapName in the local index, or "" when there is none.
|
||||||
|
func (v *Vaultik) previousSnapshotID(snapName string) (string, error) {
|
||||||
|
snapshots, err := v.Repositories.Snapshots.ListRecent(v.ctx, listRecentLimit)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("listing snapshots: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, s := range snapshots {
|
||||||
|
if s.CompletedAt != nil &&
|
||||||
|
parseSnapshotName(s.ID.String(), s.Hostname.String()) == snapName {
|
||||||
|
return s.ID.String(), nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
// resolveSnapshotPaths resolves source directories to absolute paths
|
// resolveSnapshotPaths resolves source directories to absolute paths
|
||||||
// with symlink resolution.
|
// with symlink resolution.
|
||||||
func (v *Vaultik) resolveSnapshotPaths(snapName string) ([]string, error) {
|
func (v *Vaultik) resolveSnapshotPaths(snapName string) ([]string, error) {
|
||||||
|
|||||||
@@ -0,0 +1,196 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// These tests cover https://git.eeqj.de/sneak/vaultik/issues/304: a
|
||||||
|
// backup counts a file as deleted only when the previous snapshot of
|
||||||
|
// the same name had it and it is gone, and never counts an excluded
|
||||||
|
// file.
|
||||||
|
|
||||||
|
// deletedCountEnv is a backup setup over the source directory src whose
|
||||||
|
// user-facing output is kept in out. Each test configures its snapshots
|
||||||
|
// with setSnapshot.
|
||||||
|
type deletedCountEnv struct {
|
||||||
|
v *vaultik.Vaultik
|
||||||
|
out *bytes.Buffer
|
||||||
|
src string
|
||||||
|
}
|
||||||
|
|
||||||
|
// newDeletedCountEnv writes files, keyed by path relative to src.
|
||||||
|
func newDeletedCountEnv(t *testing.T, files map[string][]byte) *deletedCountEnv {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
tempDir := t.TempDir()
|
||||||
|
srcDir := filepath.Join(tempDir, "src")
|
||||||
|
|
||||||
|
for rel, content := range files {
|
||||||
|
path := filepath.Join(srcDir, rel)
|
||||||
|
require.NoError(t, fs.MkdirAll(filepath.Dir(path), 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs, path, content, 0o644))
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg := faultTestConfig()
|
||||||
|
cfg.IndexPath = filepath.Join(tempDir, "index.sqlite")
|
||||||
|
cfg.Snapshots = map[string]config.SnapshotConfig{}
|
||||||
|
|
||||||
|
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, cfg.IndexPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
out := &bytes.Buffer{}
|
||||||
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||||
|
v.UI = ui.NewWithColor(out, false)
|
||||||
|
|
||||||
|
return &deletedCountEnv{v: v, out: out, src: srcDir}
|
||||||
|
}
|
||||||
|
|
||||||
|
// setSnapshot configures the snapshot name to back up the path rel
|
||||||
|
// beneath src, with the given excludes.
|
||||||
|
func (e *deletedCountEnv) setSnapshot(name, rel string, exclude ...string) {
|
||||||
|
e.v.Config.Snapshots[name] = config.SnapshotConfig{
|
||||||
|
Paths: []string{filepath.Join(e.src, rel)},
|
||||||
|
Exclude: exclude,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// backUp runs a backup of the named snapshot and returns its output.
|
||||||
|
func (e *deletedCountEnv) backUp(t *testing.T, name string) string {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
e.out.Reset()
|
||||||
|
require.NoError(t, e.v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
Snapshots: []string{name},
|
||||||
|
}))
|
||||||
|
|
||||||
|
return e.out.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
// A file this snapshot excludes is not counted as deleted: neither one
|
||||||
|
// that another snapshot backs up, nor one that the previous snapshot of
|
||||||
|
// this name had before the exclude was added.
|
||||||
|
func TestBackupDoesNotCountExcludedFileAsDeleted(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newDeletedCountEnv(t, map[string][]byte{
|
||||||
|
"top.txt": []byte("top"),
|
||||||
|
"cache/tmp.bin": []byte("cache"),
|
||||||
|
"photos/a.jpg": []byte("photo a"),
|
||||||
|
"photos/b.jpg": []byte("photo b"),
|
||||||
|
"photos/c.jpg": []byte("photo c"),
|
||||||
|
})
|
||||||
|
env.setSnapshot("photos", "photos")
|
||||||
|
env.setSnapshot("home", "", "/photos")
|
||||||
|
|
||||||
|
env.backUp(t, "photos")
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(2, 2, 0))
|
||||||
|
|
||||||
|
env.setSnapshot("home", "", "/photos", "/cache")
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
|
||||||
|
}
|
||||||
|
|
||||||
|
// A file inside a directory that an exclude added after a backup
|
||||||
|
// matches is not counted as deleted, though the pattern matches only the
|
||||||
|
// directory.
|
||||||
|
func TestBackupDoesNotCountFileInExcludedDirectoryAsDeleted(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newDeletedCountEnv(t, map[string][]byte{
|
||||||
|
"top.txt": []byte("top"),
|
||||||
|
"user/.local/share/Trash/old.txt": []byte("old"),
|
||||||
|
})
|
||||||
|
env.setSnapshot("home", "")
|
||||||
|
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(2, 2, 0))
|
||||||
|
|
||||||
|
env.setSnapshot("home", "", "*/.local/share/Trash")
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
|
||||||
|
}
|
||||||
|
|
||||||
|
// A backup counts only files that the previous snapshot of its own name
|
||||||
|
// had, not one that a later snapshot of another name over the same path
|
||||||
|
// had and that is since removed.
|
||||||
|
func TestBackupIgnoresSnapshotOfAnotherName(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newDeletedCountEnv(t, map[string][]byte{
|
||||||
|
"kept.txt": []byte("kept"),
|
||||||
|
})
|
||||||
|
env.setSnapshot("home", "")
|
||||||
|
env.setSnapshot("other", "")
|
||||||
|
|
||||||
|
env.backUp(t, "home")
|
||||||
|
|
||||||
|
// Snapshot start times are kept in whole seconds: wait so that other's
|
||||||
|
// snapshot is strictly the most recent.
|
||||||
|
time.Sleep(time.Second)
|
||||||
|
|
||||||
|
extra := filepath.Join(env.src, "extra.txt")
|
||||||
|
require.NoError(t, os.WriteFile(extra, []byte("extra"), 0o600))
|
||||||
|
env.backUp(t, "other")
|
||||||
|
require.NoError(t, os.Remove(extra))
|
||||||
|
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
|
||||||
|
}
|
||||||
|
|
||||||
|
// A file removed from disk is counted as deleted by the next backup of
|
||||||
|
// the snapshot that had it, and not by any later backup of that
|
||||||
|
// snapshot, nor by the first backup of another name over the same path.
|
||||||
|
func TestBackupCountsDeletedFileOnce(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
aContent := bytes.Repeat([]byte("a"), 1000)
|
||||||
|
bContent := bytes.Repeat([]byte("b"), 2000)
|
||||||
|
keptContent := bytes.Repeat([]byte("k"), 500)
|
||||||
|
|
||||||
|
env := newDeletedCountEnv(t, map[string][]byte{
|
||||||
|
"a.txt": aContent,
|
||||||
|
"b.txt": bContent,
|
||||||
|
"kept.txt": keptContent,
|
||||||
|
})
|
||||||
|
env.setSnapshot("home", "")
|
||||||
|
env.setSnapshot("other", "")
|
||||||
|
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(3, 3, 0))
|
||||||
|
|
||||||
|
require.NoError(t, os.Remove(filepath.Join(env.src, "a.txt")))
|
||||||
|
require.NoError(t, os.Remove(filepath.Join(env.src, "b.txt")))
|
||||||
|
|
||||||
|
summary := env.backUp(t, "home")
|
||||||
|
assert.Contains(t, summary,
|
||||||
|
"Files: 1 examined, 0 backed up, 1 unchanged, 2 deleted.")
|
||||||
|
assert.Contains(t, summary,
|
||||||
|
fmt.Sprintf("Data: %s total (%s backed up), %s deleted.",
|
||||||
|
env.v.UI.Size(int64(len(keptContent))), env.v.UI.Size(0),
|
||||||
|
env.v.UI.Size(int64(len(aContent)+len(bContent)))))
|
||||||
|
|
||||||
|
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
|
||||||
|
assert.Contains(t, env.backUp(t, "other"), filesLine(1, 0, 1))
|
||||||
|
}
|
||||||
@@ -1,147 +0,0 @@
|
|||||||
package vaultik_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"io/fs"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TestPruneDeletesNothingThroughSymlinkedDirectory checks that prune keeps
|
|
||||||
// every blob and the local snapshot record when the metadata/ or blobs/
|
|
||||||
// directory of a file:// destination is a symlink to a directory elsewhere
|
|
||||||
// (https://git.eeqj.de/sneak/vaultik/issues/296). A backup writes through
|
|
||||||
// such a link, so prune has to list through it too.
|
|
||||||
func TestPruneDeletesNothingThroughSymlinkedDirectory(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
for _, dir := range []string{"metadata", "blobs"} {
|
|
||||||
t.Run(dir, func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "store")
|
|
||||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
|
||||||
blobFiles := blobFilesIn(t, storeDir)
|
|
||||||
moveBehindSymlink(t, storeDir, dir)
|
|
||||||
|
|
||||||
require.NoError(t, v.Prune(&vaultik.PruneOptions{Force: true}))
|
|
||||||
|
|
||||||
// When blobs/ is the link, these paths run through it.
|
|
||||||
for _, blobFile := range blobFiles {
|
|
||||||
assert.FileExists(t, blobFile, "prune must delete no blob")
|
|
||||||
}
|
|
||||||
|
|
||||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPruneFailsOnSymlinkedDirectoryWithMissingTarget checks that prune
|
|
||||||
// fails, deleting no blob and keeping the local snapshot record, when the
|
|
||||||
// metadata/ or blobs/ directory of a file:// destination is a symlink
|
|
||||||
// whose target is missing, as a link to a volume that is not mounted is.
|
|
||||||
func TestPruneFailsOnSymlinkedDirectoryWithMissingTarget(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
for _, dir := range []string{"metadata", "blobs"} {
|
|
||||||
t.Run(dir, func(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "store")
|
|
||||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
|
||||||
blobFiles := blobFilesIn(t, storeDir)
|
|
||||||
targetPath := moveBehindSymlink(t, storeDir, dir)
|
|
||||||
|
|
||||||
// Unmount the volume the link points to, and mount it again
|
|
||||||
// once prune has run.
|
|
||||||
unmountedPath := targetPath + "-unmounted"
|
|
||||||
require.NoError(t, os.Rename(targetPath, unmountedPath))
|
|
||||||
|
|
||||||
err := v.Prune(&vaultik.PruneOptions{Force: true})
|
|
||||||
require.ErrorIs(t, err, fs.ErrNotExist)
|
|
||||||
|
|
||||||
require.NoError(t, os.Rename(unmountedPath, targetPath))
|
|
||||||
|
|
||||||
// When blobs/ is the link, these paths run through it.
|
|
||||||
for _, blobFile := range blobFiles {
|
|
||||||
assert.FileExists(t, blobFile, "prune must delete no blob")
|
|
||||||
}
|
|
||||||
|
|
||||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget checks that
|
|
||||||
// snapshot remove warns that the metadata could not be removed from the
|
|
||||||
// destination when metadata/ is a symlink whose target is missing, as it
|
|
||||||
// does when the destination directory is missing, instead of finding no
|
|
||||||
// metadata to remove.
|
|
||||||
func TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget(t *testing.T) {
|
|
||||||
log.Initialize(log.Config{})
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
storeDir := filepath.Join(t.TempDir(), "store")
|
|
||||||
v, repos, out := backUpToFileDestination(ctx, t, storeDir)
|
|
||||||
targetPath := moveBehindSymlink(t, storeDir, "metadata")
|
|
||||||
require.NoError(t, os.Rename(targetPath, targetPath+"-unmounted"))
|
|
||||||
|
|
||||||
result, err := v.RemoveSnapshot(localSnapshotID(ctx, t, repos, "first"),
|
|
||||||
&vaultik.RemoveOptions{Force: true})
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
assert.False(t, result.RemoteRemoved)
|
|
||||||
assert.Contains(t, out.String(),
|
|
||||||
"Could not remove snapshot metadata from remote")
|
|
||||||
assert.NotContains(t, out.String(),
|
|
||||||
"Removed snapshot metadata from remote storage")
|
|
||||||
}
|
|
||||||
|
|
||||||
// blobFilesIn returns the paths of the blob files under storeDir/blobs, and
|
|
||||||
// fails the test when there are none.
|
|
||||||
func blobFilesIn(t *testing.T, storeDir string) []string {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var blobFiles []string
|
|
||||||
|
|
||||||
err := filepath.WalkDir(filepath.Join(storeDir, "blobs"),
|
|
||||||
func(path string, entry fs.DirEntry, err error) error {
|
|
||||||
if err == nil && !entry.IsDir() {
|
|
||||||
blobFiles = append(blobFiles, path)
|
|
||||||
}
|
|
||||||
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NotEmpty(t, blobFiles)
|
|
||||||
|
|
||||||
return blobFiles
|
|
||||||
}
|
|
||||||
|
|
||||||
// moveBehindSymlink moves storeDir/dir to a directory elsewhere, puts a
|
|
||||||
// symlink to it in its place, and returns where it went.
|
|
||||||
func moveBehindSymlink(t *testing.T, storeDir, dir string) string {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
linkPath := filepath.Join(storeDir, dir)
|
|
||||||
targetPath := filepath.Join(t.TempDir(), dir)
|
|
||||||
require.NoError(t, os.Rename(linkPath, targetPath))
|
|
||||||
require.NoError(t, os.Symlink(targetPath, linkPath))
|
|
||||||
|
|
||||||
return targetPath
|
|
||||||
}
|
|
||||||
+17
-39
@@ -17,7 +17,6 @@ import (
|
|||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Sentinel errors for snapshot verification failures.
|
// Sentinel errors for snapshot verification failures.
|
||||||
@@ -33,9 +32,7 @@ var (
|
|||||||
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
||||||
errManifestMissingBlob = errors.New(
|
errManifestMissingBlob = errors.New(
|
||||||
"manifest omits blob present in database")
|
"manifest omits blob present in database")
|
||||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||||
errBlobsNotInStorage = errors.New(
|
|
||||||
"blobs missing from storage or stored with the wrong size")
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||||
@@ -152,6 +149,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
result.Status = "ok"
|
result.Status = "ok"
|
||||||
|
result.Verified = len(dbBlobs)
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputVerifyJSON(result)
|
return v.outputVerifyJSON(result)
|
||||||
@@ -308,7 +306,7 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -319,8 +317,7 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
len(dbBlobs), ubytes(totalSize))
|
len(dbBlobs), ubytes(totalSize))
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.performDeepVerificationFromDB(
|
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
||||||
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -672,14 +669,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// verifyBlobExistenceFromDB checks that every blob from the database is
|
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
||||||
// in storage with the size the database records. It checks them all,
|
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
||||||
// counting the missing and wrong-size blobs in result, and fails if it
|
|
||||||
// found any. Any storage error other than not found stops the check and
|
|
||||||
// fails with that error.
|
|
||||||
func (v *Vaultik) verifyBlobExistenceFromDB(
|
|
||||||
blobs []snapshot.BlobInfo, result *VerifyResult,
|
|
||||||
) error {
|
|
||||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||||
|
|
||||||
for i, blob := range blobs {
|
for i, blob := range blobs {
|
||||||
@@ -692,23 +683,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
|||||||
// Construct blob path
|
// Construct blob path
|
||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
||||||
|
|
||||||
|
// Check blob exists
|
||||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||||
switch {
|
if err != nil {
|
||||||
case errors.Is(err, storage.ErrNotFound):
|
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
||||||
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
}
|
||||||
|
|
||||||
result.Missing++
|
// Verify size matches
|
||||||
result.MissingSize += blob.CompressedSize
|
if stat.Size != blob.CompressedSize {
|
||||||
case err != nil:
|
return fmt.Errorf(
|
||||||
return fmt.Errorf("checking blob %s in storage: %w",
|
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
||||||
blob.Hash, err)
|
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
||||||
case stat.Size != blob.CompressedSize:
|
|
||||||
log.Warn("Blob has the wrong size in storage",
|
|
||||||
"hash", blob.Hash,
|
|
||||||
"storage_size", stat.Size,
|
|
||||||
"database_size", blob.CompressedSize)
|
|
||||||
|
|
||||||
result.Mismatched++
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Progress update every 100 blobs
|
// Progress update every 100 blobs
|
||||||
@@ -722,22 +707,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.Missing > 0 || result.Mismatched > 0 {
|
|
||||||
return fmt.Errorf("%w: %d missing, %d with the wrong size",
|
|
||||||
errBlobsNotInStorage, result.Missing, result.Mismatched)
|
|
||||||
}
|
|
||||||
|
|
||||||
log.Info("✓ All blobs exist in storage")
|
log.Info("✓ All blobs exist in storage")
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// performDeepVerificationFromDB downloads and verifies the content of
|
// performDeepVerificationFromDB downloads and verifies the content of
|
||||||
// each blob using the database as source, stopping at the first blob that
|
// each blob using the database as source.
|
||||||
// fails. Each blob that passes is counted in result.Verified.
|
|
||||||
func (v *Vaultik) performDeepVerificationFromDB(
|
func (v *Vaultik) performDeepVerificationFromDB(
|
||||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||||
result *VerifyResult, identities []age.Identity,
|
identities []age.Identity,
|
||||||
) error {
|
) error {
|
||||||
// Calculate total bytes for ETA
|
// Calculate total bytes for ETA
|
||||||
var totalBytesExpected int64
|
var totalBytesExpected int64
|
||||||
@@ -760,7 +739,6 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
|||||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
result.Verified++
|
|
||||||
bytesProcessed += blobInfo.CompressedSize
|
bytesProcessed += blobInfo.CompressedSize
|
||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
remaining := len(blobs) - (i + 1)
|
remaining := len(blobs) - (i + 1)
|
||||||
|
|||||||
Reference in New Issue
Block a user