Compare commits
6
Commits
c03d988c3d
...
3fd08c164d
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3fd08c164d | ||
|
|
fb8abcd049 | ||
|
|
6aceaaf086 | ||
|
|
3b2b9cec84 | ||
|
|
dfa3bbc9e1 | ||
|
|
740f474c59 |
No files matched your search
+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()
|
||||||
|
|||||||
@@ -375,7 +375,9 @@ on the destination in one go, use `vaultik remote nuke --force`.
|
|||||||
* Optional path arguments to restore specific files/directories (default: all)
|
* Optional path arguments to restore specific files/directories (default: all)
|
||||||
* Preserves file permissions, timestamps, ownership (ownership requires root),
|
* Preserves file permissions, timestamps, ownership (ownership requires root),
|
||||||
symlinks, and empty directories
|
symlinks, and empty directories
|
||||||
* `--verify`: After restoring, verify every file's chunk hashes match
|
* `--verify`: After restoring, verify every restored file's chunk hashes
|
||||||
|
match. A file skipped under `--skip-errors` is reported as failed to
|
||||||
|
restore and is not verified.
|
||||||
|
|
||||||
**`prune`**: Tidy up everything that isn't needed. Runs three passes:
|
**`prune`**: Tidy up everything that isn't needed. Runs three passes:
|
||||||
(1) reconcile the local index against the destination store — any
|
(1) reconcile the local index against the destination store — any
|
||||||
@@ -386,10 +388,12 @@ manifest on the destination store to compute the still-referenced blob
|
|||||||
set and delete any blob not in that set. Step (3) reads all remote
|
set and delete any blob not in that set. Step (3) reads all remote
|
||||||
manifests — network cost scales with the number of snapshots. `snapshot
|
manifests — network cost scales with the number of snapshots. `snapshot
|
||||||
create --prune` runs the same cleanup automatically; this is the
|
create --prune` runs the same cleanup automatically; this is the
|
||||||
manual entry point for the same work.
|
manual entry point for the same work. If a blob cannot be deleted, the
|
||||||
|
command (and `snapshot create --prune`) exits non-zero.
|
||||||
* `--force`: Skip confirmation prompt
|
* `--force`: Skip confirmation prompt
|
||||||
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
||||||
`--force` does.
|
`--force` does. A blob that cannot be deleted is counted in
|
||||||
|
`blobs_failed`, and the document is printed before the non-zero exit.
|
||||||
|
|
||||||
**`info`**: Display system configuration, storage settings, encryption
|
**`info`**: Display system configuration, storage settings, encryption
|
||||||
recipients, and local database statistics.
|
recipients, and local database statistics.
|
||||||
@@ -411,7 +415,8 @@ since `prune` never deletes it.
|
|||||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||||
backup destination store, leaving the bucket prefix empty. Destructive and
|
backup destination store, leaving the bucket prefix empty. Destructive and
|
||||||
irreversible. This is the single supported way to wipe the entire
|
irreversible. This is the single supported way to wipe the entire
|
||||||
destination store.
|
destination store. If anything cannot be deleted, the command exits
|
||||||
|
non-zero; running it again deletes what is left.
|
||||||
* `--force`: Required to confirm destruction.
|
* `--force`: Required to confirm destruction.
|
||||||
|
|
||||||
**`database delete`**: Delete the local SQLite state database file
|
**`database delete`**: Delete the local SQLite state database file
|
||||||
|
|||||||
+2
-2
@@ -90,7 +90,7 @@ Stores information about packed, compressed, and encrypted blob files.
|
|||||||
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
||||||
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
||||||
- `uncompressed_size` (INTEGER NOT NULL DEFAULT 0) - Total size of chunks before compression
|
- `uncompressed_size` (INTEGER NOT NULL DEFAULT 0) - Total size of chunks before compression
|
||||||
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption
|
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption. Once uploaded, the size of the object in storage: when the blob was already stored, under another compression level or other recipients, this can differ from the size of the local copy
|
||||||
- `uploaded_ts` (INTEGER) - Upload completion timestamp (NULL if not uploaded)
|
- `uploaded_ts` (INTEGER) - Upload completion timestamp (NULL if not uploaded)
|
||||||
|
|
||||||
**Purpose:** Tracks blob lifecycle from creation through upload. The UUID primary key allows immediate association of chunks with blobs.
|
**Purpose:** Tracks blob lifecycle from creation through upload. The UUID primary key allows immediate association of chunks with blobs.
|
||||||
@@ -184,7 +184,7 @@ Tracks blob upload metrics.
|
|||||||
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
||||||
|
|
||||||
4. **Upload**
|
4. **Upload**
|
||||||
- `UPDATE blobs SET uploaded_ts = ?` - Mark blob as uploaded
|
- `UPDATE blobs SET uploaded_ts = ?, compressed_size = ?` - Mark blob as uploaded, with the size of the object in storage (the upload is skipped when the blob is already stored)
|
||||||
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
||||||
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
||||||
|
|
||||||
|
|||||||
@@ -58,12 +58,7 @@ func (r *BlobChunkRepository) GetByBlobID(
|
|||||||
return nil, fmt.Errorf("querying blob chunks: %w", err)
|
return nil, fmt.Errorf("querying blob chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var blobChunks []*BlobChunk
|
var blobChunks []*BlobChunk
|
||||||
|
|
||||||
|
|||||||
@@ -83,12 +83,7 @@ func (r *BlobRepository) GetAll(ctx context.Context) (map[string]*Blob, error) {
|
|||||||
return nil, fmt.Errorf("querying blobs: %w", err)
|
return nil, fmt.Errorf("querying blobs: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
out := make(map[string]*Blob)
|
out := make(map[string]*Blob)
|
||||||
|
|
||||||
@@ -159,13 +154,16 @@ func (r *BlobRepository) UpdateFinished(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateUploaded marks a blob as uploaded
|
// UpdateUploaded marks a blob as uploaded and records the size of the
|
||||||
|
// stored object as its compressed size. That size can differ from the
|
||||||
|
// local copy's when an identical blob was already stored with another
|
||||||
|
// compression level or other recipients.
|
||||||
func (r *BlobRepository) UpdateUploaded(
|
func (r *BlobRepository) UpdateUploaded(
|
||||||
ctx context.Context, tx *sql.Tx, id string,
|
ctx context.Context, tx *sql.Tx, id string, storedSize int64,
|
||||||
) error {
|
) error {
|
||||||
query := `
|
query := `
|
||||||
UPDATE blobs
|
UPDATE blobs
|
||||||
SET uploaded_ts = ?
|
SET uploaded_ts = ?, compressed_size = ?
|
||||||
WHERE id = ?
|
WHERE id = ?
|
||||||
`
|
`
|
||||||
|
|
||||||
@@ -173,9 +171,9 @@ func (r *BlobRepository) UpdateUploaded(
|
|||||||
|
|
||||||
var err error
|
var err error
|
||||||
if tx != nil {
|
if tx != nil {
|
||||||
_, err = tx.ExecContext(ctx, query, now, id)
|
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
||||||
} else {
|
} else {
|
||||||
_, err = r.db.ExecWithLog(ctx, query, now, id)
|
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -122,8 +122,8 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
|||||||
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Test UpdateUploaded
|
// Test UpdateUploaded, with a stored object larger than the local copy
|
||||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("failed to update blob as uploaded: %v", err)
|
t.Fatalf("failed to update blob as uploaded: %v", err)
|
||||||
}
|
}
|
||||||
@@ -141,6 +141,10 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
|||||||
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
||||||
t.Error("uploaded timestamp should be around test time")
|
t.Error("uploaded timestamp should be around test time")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if uploaded.CompressedSize != 600 {
|
||||||
|
t.Errorf("expected the stored size 600, got %d", uploaded.CompressedSize)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestBlobRepositoryDuplicate(t *testing.T) {
|
func TestBlobRepositoryDuplicate(t *testing.T) {
|
||||||
|
|||||||
@@ -61,12 +61,7 @@ func (r *ChunkFileRepository) GetByChunkHash(
|
|||||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanChunkFiles(rows)
|
return r.scanChunkFiles(rows)
|
||||||
}
|
}
|
||||||
@@ -87,12 +82,7 @@ func (r *ChunkFileRepository) GetByFilePath(
|
|||||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanChunkFiles(rows)
|
return r.scanChunkFiles(rows)
|
||||||
}
|
}
|
||||||
@@ -112,12 +102,7 @@ func (r *ChunkFileRepository) GetByFileID(
|
|||||||
return nil, fmt.Errorf("querying chunk files: %w", err)
|
return nil, fmt.Errorf("querying chunk files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanChunkFiles(rows)
|
return r.scanChunkFiles(rows)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -107,12 +107,7 @@ func (r *ChunkRepository) GetByHashes(
|
|||||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var chunks []*Chunk
|
var chunks []*Chunk
|
||||||
|
|
||||||
@@ -152,12 +147,7 @@ func (r *ChunkRepository) ListUnpacked(
|
|||||||
return nil, fmt.Errorf("querying unpacked chunks: %w", err)
|
return nil, fmt.Errorf("querying unpacked chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var chunks []*Chunk
|
var chunks []*Chunk
|
||||||
|
|
||||||
|
|||||||
@@ -38,12 +38,7 @@ func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, err
|
|||||||
return nil, fmt.Errorf("querying chunks: %w", err)
|
return nil, fmt.Errorf("querying chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var chunks []*Chunk
|
var chunks []*Chunk
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +0,0 @@
|
|||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
)
|
|
||||||
|
|
||||||
// Fatalf prints an error message to stderr and exits with status 1
|
|
||||||
func Fatalf(format string, args ...any) {
|
|
||||||
fmt.Fprintf(os.Stderr, "FATAL: "+format+"\n", args...)
|
|
||||||
os.Exit(1)
|
|
||||||
}
|
|
||||||
@@ -62,12 +62,7 @@ func (r *FileChunkRepository) GetByPath(
|
|||||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanFileChunks(rows)
|
return r.scanFileChunks(rows)
|
||||||
}
|
}
|
||||||
@@ -88,12 +83,7 @@ func (r *FileChunkRepository) GetByFileID(
|
|||||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanFileChunks(rows)
|
return r.scanFileChunks(rows)
|
||||||
}
|
}
|
||||||
@@ -117,12 +107,7 @@ func (r *FileChunkRepository) GetByPathTx(
|
|||||||
return nil, fmt.Errorf("querying file chunks: %w", err)
|
return nil, fmt.Errorf("querying file chunks: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
fileChunks, err := r.scanFileChunks(rows)
|
fileChunks, err := r.scanFileChunks(rows)
|
||||||
LogSQL("GetByPathTx", "Complete", path, "count", len(fileChunks))
|
LogSQL("GetByPathTx", "Complete", path, "count", len(fileChunks))
|
||||||
|
|||||||
+41
-22
@@ -175,12 +175,7 @@ func (r *FileRepository) ListModifiedSince(
|
|||||||
return nil, fmt.Errorf("querying files: %w", err)
|
return nil, fmt.Errorf("querying files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var files []*File
|
var files []*File
|
||||||
|
|
||||||
@@ -256,12 +251,46 @@ func (r *FileRepository) ListUnderPath(
|
|||||||
return nil, fmt.Errorf("querying files: %w", err)
|
return nil, fmt.Errorf("querying files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
|
var files []*File
|
||||||
|
|
||||||
|
for rows.Next() {
|
||||||
|
file, err := r.scanFileRows(rows)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
Fatalf("failed to close rows: %v", err)
|
return nil, fmt.Errorf("scanning file: %w", err)
|
||||||
}
|
}
|
||||||
}()
|
|
||||||
|
files = append(files, file)
|
||||||
|
}
|
||||||
|
|
||||||
|
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
|
var files []*File
|
||||||
|
|
||||||
@@ -304,12 +333,7 @@ func (r *FileRepository) ListIDsWithChunksNotInUploadedBlobs(
|
|||||||
return nil, fmt.Errorf("querying files: %w", err)
|
return nil, fmt.Errorf("querying files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var ids []types.FileID
|
var ids []types.FileID
|
||||||
|
|
||||||
@@ -340,12 +364,7 @@ func (r *FileRepository) ListAll(ctx context.Context) ([]*File, error) {
|
|||||||
return nil, fmt.Errorf("querying files: %w", err)
|
return nil, fmt.Errorf("querying files: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var files []*File
|
var files []*File
|
||||||
|
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|
||||||
|
|||||||
@@ -106,22 +106,24 @@ func (r *Repositories) WithReadTx(ctx context.Context, fn TxFunc) error {
|
|||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if p := recover(); p != nil {
|
if p := recover(); p != nil {
|
||||||
rollbackErr := tx.Rollback()
|
// The panic is the failure to report, so a rollback error
|
||||||
if rollbackErr != nil {
|
// is dropped.
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
_ = tx.Rollback()
|
||||||
}
|
|
||||||
|
|
||||||
panic(p)
|
panic(p)
|
||||||
} else if err != nil {
|
|
||||||
rollbackErr := tx.Rollback()
|
|
||||||
if rollbackErr != nil {
|
|
||||||
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
err = fn(ctx, tx)
|
err = fn(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// Once ctx is cancelled, database/sql rolls the transaction
|
||||||
|
// back itself and Rollback returns sql.ErrTxDone.
|
||||||
|
rollbackErr := tx.Rollback()
|
||||||
|
if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
|
||||||
|
return fmt.Errorf("%w; rolling back transaction: %w",
|
||||||
|
err, rollbackErr)
|
||||||
|
}
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -411,3 +411,41 @@ func TestRepositoriesReadTransaction(t *testing.T) {
|
|||||||
t.Error("expected to retrieve file in read transaction")
|
t.Error("expected to retrieve file in read transaction")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// When the context is cancelled during a read transaction's body,
|
||||||
|
// database/sql rolls the transaction back itself, so WithReadTx's own
|
||||||
|
// rollback finds it already done. WithReadTx must return the body's error
|
||||||
|
// alone, without that rollback.
|
||||||
|
func TestRepositoriesReadTransactionContextCancelled(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
db, cleanup := setupTestDB(t)
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
err := repos.WithReadTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
// database/sql rolls back from a goroutine of its own; wait
|
||||||
|
// until it has, so the test does not depend on that race.
|
||||||
|
for {
|
||||||
|
_, err := tx.ExecContext(context.WithoutCancel(ctx), "SELECT 1")
|
||||||
|
if errors.Is(err, sql.ErrTxDone) {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(time.Millisecond)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("expected context.Canceled, got: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if errors.Is(err, sql.ErrTxDone) {
|
||||||
|
t.Fatalf("the already-done rollback was reported: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -241,12 +241,7 @@ func (r *SnapshotRepository) GetOnlySnapshot(ctx context.Context) (*Snapshot, er
|
|||||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
snapshots, err := r.scanSnapshotRows(rows)
|
snapshots, err := r.scanSnapshotRows(rows)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -281,12 +276,7 @@ func (r *SnapshotRepository) ListRecent(
|
|||||||
return nil, fmt.Errorf("querying snapshots: %w", err)
|
return nil, fmt.Errorf("querying snapshots: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanSnapshotRows(rows)
|
return r.scanSnapshotRows(rows)
|
||||||
}
|
}
|
||||||
@@ -501,12 +491,7 @@ func (r *SnapshotRepository) GetBlobHashes(
|
|||||||
return nil, fmt.Errorf("querying blob hashes: %w", err)
|
return nil, fmt.Errorf("querying blob hashes: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var blobs []string
|
var blobs []string
|
||||||
|
|
||||||
@@ -634,12 +619,7 @@ func (r *SnapshotRepository) GetIncompleteSnapshots(
|
|||||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
return r.scanSnapshotRows(rows)
|
return r.scanSnapshotRows(rows)
|
||||||
}
|
}
|
||||||
@@ -662,12 +642,7 @@ func (r *SnapshotRepository) GetIncompleteByHostname(
|
|||||||
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
return nil, fmt.Errorf("querying incomplete snapshots: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
err := rows.Close()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Same column set as every other multi-row snapshot query, so the
|
// Same column set as every other multi-row snapshot query, so the
|
||||||
// shared scanner applies — including its timestamp normalization.
|
// shared scanner applies — including its timestamp normalization.
|
||||||
|
|||||||
@@ -0,0 +1,62 @@
|
|||||||
|
//nolint:testpackage // exercises the unexported cleanupBlobTempFile
|
||||||
|
package snapshot
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"sneak.berlin/go/vaultik/internal/blob"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A blob temp file that cannot be closed or removed must make the cleanup
|
||||||
|
// return an error, not end the process.
|
||||||
|
func TestCleanupBlobTempFileReturnsErrors(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
s := &Scanner{fs: fs}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
// prepare runs before the cleanup and makes its close or remove
|
||||||
|
// fail.
|
||||||
|
prepare func(tempFile afero.File) error
|
||||||
|
want error
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "close fails",
|
||||||
|
prepare: func(tempFile afero.File) error { return tempFile.Close() },
|
||||||
|
want: os.ErrClosed,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "remove fails",
|
||||||
|
prepare: func(tempFile afero.File) error {
|
||||||
|
return fs.Remove(tempFile.Name())
|
||||||
|
},
|
||||||
|
want: os.ErrNotExist,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tempFile, err := afero.TempFile(fs, t.TempDir(), "blob-*.tmp")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = tt.prepare(tempFile)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.cleanupBlobTempFile(&blob.WithReader{TempFile: tempFile})
|
||||||
|
if !errors.Is(err, tt.want) {
|
||||||
|
t.Fatalf("expected an error wrapping %v, got: %v", tt.want, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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.
|
||||||
@@ -55,6 +58,7 @@ func provideScannerFactory(
|
|||||||
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) {
|
||||||
@@ -1517,7 +1525,8 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
|
|||||||
// snapshot references has uploaded_ts set, so deduplication and
|
// snapshot references has uploaded_ts set, so deduplication and
|
||||||
// interrupted-run repair treat these blobs as trustworthy.
|
// interrupted-run repair treat these blobs as trustworthy.
|
||||||
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
err := s.repos.Blobs.UpdateUploaded(
|
||||||
|
ctx, tx, b.ID, b.Compressed)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||||
}
|
}
|
||||||
@@ -1553,23 +1562,28 @@ func (s *Scanner) handleBlobReady(
|
|||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
||||||
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
||||||
|
|
||||||
blobExists, err := s.uploadBlobIfNeeded(
|
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
||||||
ctx, blobPath, blobWithReader, startTime, result)
|
ctx, blobPath, blobWithReader, startTime, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.cleanupBlobTempFile(blobWithReader)
|
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||||
|
|
||||||
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
return errors.Join(
|
||||||
|
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
err = s.recordBlobMetadata(
|
||||||
|
ctx, finishedBlob, storedSize, blobExists, startTime)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.cleanupBlobTempFile(blobWithReader)
|
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||||
|
|
||||||
|
return errors.Join(err, cleanupErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.cleanupBlobTempFile(blobWithReader)
|
||||||
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
s.cleanupBlobTempFile(blobWithReader)
|
|
||||||
|
|
||||||
// Chunks from this blob are now committed to DB - remove from pending set
|
// Chunks from this blob are now committed to DB - remove from pending set
|
||||||
s.removePendingChunkHashes(blobWithReader.InsertedChunkHashes)
|
s.removePendingChunkHashes(blobWithReader.InsertedChunkHashes)
|
||||||
|
|
||||||
@@ -1583,20 +1597,22 @@ func (s *Scanner) handleBlobReady(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
||||||
// exist, returns whether it existed
|
// exist. It returns the size of the object in storage, which for an
|
||||||
|
// existing object can differ from the local copy's, and whether it
|
||||||
|
// already existed.
|
||||||
func (s *Scanner) uploadBlobIfNeeded(
|
func (s *Scanner) uploadBlobIfNeeded(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
blobPath string,
|
blobPath string,
|
||||||
blobWithReader *blob.WithReader,
|
blobWithReader *blob.WithReader,
|
||||||
startTime time.Time,
|
startTime time.Time,
|
||||||
result *ScanResult,
|
result *ScanResult,
|
||||||
) (bool, error) {
|
) (int64, bool, error) {
|
||||||
finishedBlob := blobWithReader.FinishedBlob
|
finishedBlob := blobWithReader.FinishedBlob
|
||||||
|
|
||||||
// Check if blob already exists (deduplication after restart)
|
// Check if blob already exists (deduplication after restart)
|
||||||
destination := s.storage.Info().Location
|
destination := s.storage.Info().Location
|
||||||
|
|
||||||
_, err := s.storage.Stat(ctx, blobPath)
|
stored, err := s.storage.Stat(ctx, blobPath)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
log.Info("Blob already exists in storage, skipping upload",
|
log.Info("Blob already exists in storage, skipping upload",
|
||||||
"hash", finishedBlob.Hash,
|
"hash", finishedBlob.Hash,
|
||||||
@@ -1605,7 +1621,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
||||||
s.ui.Path(destination))
|
s.ui.Path(destination))
|
||||||
|
|
||||||
return true, nil
|
return stored.Size, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||||
@@ -1619,7 +1635,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
||||||
|
|
||||||
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
uploadDuration := time.Since(startTime)
|
uploadDuration := time.Since(startTime)
|
||||||
@@ -1650,7 +1666,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||||
}
|
}
|
||||||
|
|
||||||
return false, nil
|
return finishedBlob.Compressed, false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||||
@@ -1720,6 +1736,7 @@ func (s *Scanner) makeUploadProgressCallback(
|
|||||||
func (s *Scanner) recordBlobMetadata(
|
func (s *Scanner) recordBlobMetadata(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
finishedBlob *blob.FinishedBlob,
|
finishedBlob *blob.FinishedBlob,
|
||||||
|
storedSize int64,
|
||||||
blobExists bool,
|
blobExists bool,
|
||||||
startTime time.Time,
|
startTime time.Time,
|
||||||
) error {
|
) error {
|
||||||
@@ -1731,7 +1748,8 @@ func (s *Scanner) recordBlobMetadata(
|
|||||||
uploadDuration := time.Since(startTime)
|
uploadDuration := time.Since(startTime)
|
||||||
|
|
||||||
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
||||||
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
|
err := s.repos.Blobs.UpdateUploaded(
|
||||||
|
txCtx, tx, finishedBlob.ID, storedSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
||||||
}
|
}
|
||||||
@@ -1762,20 +1780,22 @@ func (s *Scanner) recordBlobMetadata(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// cleanupBlobTempFile closes and removes the blob's temporary file
|
// cleanupBlobTempFile closes and removes the blob's temporary file
|
||||||
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) {
|
func (s *Scanner) cleanupBlobTempFile(blobWithReader *blob.WithReader) error {
|
||||||
if blobWithReader.TempFile != nil {
|
if blobWithReader.TempFile != nil {
|
||||||
tempName := blobWithReader.TempFile.Name()
|
tempName := blobWithReader.TempFile.Name()
|
||||||
|
|
||||||
err := blobWithReader.TempFile.Close()
|
err := blobWithReader.TempFile.Close()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal("Failed to close temp file", "file", tempName, "error", err)
|
return fmt.Errorf("closing blob temp file: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = s.fs.Remove(tempName)
|
err = s.fs.Remove(tempName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal("Failed to remove temp file", "file", tempName, "error", err)
|
return fmt.Errorf("removing blob temp file: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// streamingChunkInfo tracks chunk metadata collected during streaming
|
// streamingChunkInfo tracks chunk metadata collected during streaming
|
||||||
@@ -1942,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():
|
||||||
@@ -1964,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]
|
||||||
|
if exists || s.excludedOrInExcludedDir(filePath, path) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
result.FilesDeleted++
|
result.FilesDeleted++
|
||||||
result.BytesDeleted += file.Size
|
result.BytesDeleted += file.Size
|
||||||
log.Debug("Detected deleted file", "path", path, "size", file.Size)
|
log.Debug("Detected deleted file", "path", filePath, "size", file.Size)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.FilesDeleted > 0 {
|
if result.FilesDeleted > 0 {
|
||||||
@@ -1981,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
|
||||||
|
|||||||
@@ -1,15 +1,16 @@
|
|||||||
// Package faultstore provides a storage.Storer wrapper that injects
|
// Package faultstore provides a storage.Storer wrapper that injects
|
||||||
// faults on demand, so tests can reproduce the failure modes a real
|
// faults on demand, so tests can reproduce the failure modes a real
|
||||||
// backend exhibits: an upload that fails partway, a backend that reports
|
// backend exhibits: an upload that fails partway, a backend that reports
|
||||||
// success while storing nothing, and reads that return corrupt or
|
// success while storing nothing, reads that return corrupt or truncated
|
||||||
// truncated bytes. It is the seam called for by the fault-injection
|
// bytes, and a delete the backend refuses. It is the seam called for by
|
||||||
// tests (sneak/vaultik issue 72) and is meant to be reused by future
|
// the fault-injection tests (sneak/vaultik issue 72) and is meant to be
|
||||||
// tests rather than re-implemented per case.
|
// reused by future tests rather than re-implemented per case.
|
||||||
//
|
//
|
||||||
// The wrapper delegates every method to the inner Storer. Two hooks
|
// The wrapper delegates every method to the inner Storer. Three hooks
|
||||||
// change that: OnPut decides the fate of each write, and OnGet decides
|
// change that: OnPut decides the fate of each write, OnGet decides how
|
||||||
// how each read's bytes are returned. Both are keyed by the object key,
|
// each read's bytes are returned, and OnDelete decides whether each
|
||||||
// so a test can fault only blobs, only metadata, or a single object.
|
// delete fails. All are keyed by the object key, so a test can fault
|
||||||
|
// only blobs, only metadata, or a single object.
|
||||||
package faultstore
|
package faultstore
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -25,6 +26,10 @@ import (
|
|||||||
// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
|
// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
|
||||||
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")
|
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")
|
||||||
|
|
||||||
|
// ErrInjectedDelete is returned by a Delete the OnDelete hook chose to
|
||||||
|
// fail.
|
||||||
|
var ErrInjectedDelete = errors.New("faultstore: injected delete failure")
|
||||||
|
|
||||||
// PutAction is the disposition OnPut assigns to a write.
|
// PutAction is the disposition OnPut assigns to a write.
|
||||||
type PutAction int
|
type PutAction int
|
||||||
|
|
||||||
@@ -66,6 +71,11 @@ type Storer struct {
|
|||||||
// OnGet, when set, is consulted for every Get with the object key
|
// OnGet, when set, is consulted for every Get with the object key
|
||||||
// and damages the returned bytes accordingly.
|
// and damages the returned bytes accordingly.
|
||||||
OnGet func(key string) GetFault
|
OnGet func(key string) GetFault
|
||||||
|
|
||||||
|
// OnDelete, when set, is consulted before every Delete with the
|
||||||
|
// object key. Returning true fails the delete and leaves the object
|
||||||
|
// in place.
|
||||||
|
OnDelete func(key string) bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// New wraps inner. inner must be non-nil.
|
// New wraps inner. inner must be non-nil.
|
||||||
@@ -154,8 +164,12 @@ func (f *Storer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, err
|
|||||||
return f.inner.Stat(ctx, key)
|
return f.inner.Stat(ctx, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Delete delegates unchanged.
|
// Delete removes the object unless OnDelete fails the delete.
|
||||||
func (f *Storer) Delete(ctx context.Context, key string) error {
|
func (f *Storer) Delete(ctx context.Context, key string) error {
|
||||||
|
if f.OnDelete != nil && f.OnDelete(key) {
|
||||||
|
return fmt.Errorf("%w for %q", ErrInjectedDelete, key)
|
||||||
|
}
|
||||||
|
|
||||||
return f.inner.Delete(ctx, key)
|
return f.inner.Delete(ctx, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -672,6 +672,7 @@ func faultScannerFactory(
|
|||||||
UI: params.UI,
|
UI: params.UI,
|
||||||
Exclude: params.Exclude,
|
Exclude: params.Exclude,
|
||||||
SkipErrors: params.SkipErrors,
|
SkipErrors: params.SkipErrors,
|
||||||
|
PreviousSnapshotID: params.PreviousSnapshotID,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,12 +5,14 @@ import (
|
|||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"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/faultstore"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
||||||
@@ -23,7 +25,7 @@ func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
|||||||
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
storeDir := filepath.Join(t.TempDir(), "store")
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
v, repos, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -61,4 +63,82 @@ func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
|||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Empty(t, files)
|
assert.Empty(t, files)
|
||||||
|
|
||||||
|
// Nuke deletes the blobs itself, so advice to run prune is wrong.
|
||||||
|
assert.Contains(t, out.String(), "Backup destination store is now empty.")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik prune")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteFailsWhenADeleteFails checks that remote nuke returns an
|
||||||
|
// error when a snapshot's metadata or a blob cannot be deleted, does not
|
||||||
|
// report the store empty, and advises running nuke again: prune deletes
|
||||||
|
// no metadata.
|
||||||
|
func TestNukeRemoteFailsWhenADeleteFails(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, refusedPrefix := range []string{"metadata/", "blobs/"} {
|
||||||
|
t.Run(refusedPrefix, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
store := faultstore.New(v.Storage)
|
||||||
|
store.OnDelete = func(key string) bool {
|
||||||
|
return strings.HasPrefix(key, refusedPrefix)
|
||||||
|
}
|
||||||
|
v.Storage = store
|
||||||
|
|
||||||
|
require.Error(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
assert.NotContains(t, out.String(), "now empty")
|
||||||
|
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik prune")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice checks that
|
||||||
|
// remote nuke, when the local index belongs to a different destination
|
||||||
|
// than the configured one, fails without advising to run nuke again:
|
||||||
|
// that run would fail the same way.
|
||||||
|
func TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
v.Config.StorageURL = "file://" + filepath.Join(t.TempDir(), "elsewhere")
|
||||||
|
|
||||||
|
err := v.NukeRemote(true)
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "bound to a different backup destination")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik remote nuke --force")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved checks that a
|
||||||
|
// snapshot remote nuke cannot remove from the local index fails the
|
||||||
|
// run: nuke then leaves that snapshot's metadata on the destination
|
||||||
|
// store.
|
||||||
|
func TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
_, err := v.DB.Conn().ExecContext(ctx, `
|
||||||
|
CREATE TRIGGER refuse_snapshot_delete BEFORE DELETE ON snapshots
|
||||||
|
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
require.Error(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
assert.NotContains(t, out.String(), "now empty")
|
||||||
|
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||||
}
|
}
|
||||||
@@ -19,6 +19,14 @@ type PruneOptions struct {
|
|||||||
var errNukeRequiresForce = errors.New(
|
var errNukeRequiresForce = errors.New(
|
||||||
"nuke requires --force (this deletes ALL remote snapshots and blobs)")
|
"nuke requires --force (this deletes ALL remote snapshots and blobs)")
|
||||||
|
|
||||||
|
// errBlobDeleteFailed reports unreferenced blobs prune could not delete.
|
||||||
|
var errBlobDeleteFailed = errors.New("blob(s) could not be deleted")
|
||||||
|
|
||||||
|
// nukeCommandHint is the command suggested when a remote nuke could not
|
||||||
|
// delete everything: running it again deletes what is left, which
|
||||||
|
// `vaultik prune` does not do for snapshot metadata.
|
||||||
|
const nukeCommandHint = "vaultik remote nuke --force"
|
||||||
|
|
||||||
// metadataDirName is the top-level remote directory holding snapshot
|
// metadataDirName is the top-level remote directory holding snapshot
|
||||||
// metadata.
|
// metadata.
|
||||||
const metadataDirName = "metadata"
|
const metadataDirName = "metadata"
|
||||||
@@ -39,6 +47,29 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
|||||||
return errNukeRequiresForce
|
return errNukeRequiresForce
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Checked before anything is deleted: running nuke again would fail
|
||||||
|
// the same way, so the advice below must not follow this error.
|
||||||
|
err := v.EnsureStorageBinding()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = v.emptyDestinationStore()
|
||||||
|
if err != nil {
|
||||||
|
v.UI.Warningf("Run '%s' again to delete what is left on the "+
|
||||||
|
"backup destination store.", nukeCommandHint)
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
v.UI.Completef("Backup destination store is now empty.")
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// emptyDestinationStore does the deleting for NukeRemote and stops at
|
||||||
|
// the first step that fails.
|
||||||
|
func (v *Vaultik) emptyDestinationStore() error {
|
||||||
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
||||||
|
|
||||||
_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
|
_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
|
||||||
@@ -62,8 +93,6 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
v.UI.Completef("Backup destination store is now empty.")
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -166,14 +195,17 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
|
|||||||
v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
|
v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputPruneBlobsJSON(result)
|
err = v.outputPruneBlobsJSON(result)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
||||||
result.BlobsDeleted, ubytes(result.BytesFreed))
|
result.BlobsDeleted, ubytes(result.BytesFreed))
|
||||||
|
}
|
||||||
|
|
||||||
if result.BlobsFailed > 0 {
|
if result.BlobsFailed > 0 {
|
||||||
v.stdoutf("Failed to delete %d blob(s)\n", result.BlobsFailed)
|
return fmt.Errorf("%d %w", result.BlobsFailed, errBlobDeleteFailed)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -0,0 +1,98 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// refuseBlobDeletes is an OnDelete hook that fails every blob delete.
|
||||||
|
func refuseBlobDeletes(key string) bool {
|
||||||
|
return strings.HasPrefix(key, "blobs/")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPruneBlobs_FailedDeleteReturnsError checks that prune returns an
|
||||||
|
// error, so the command exits non-zero, when an unreferenced blob cannot
|
||||||
|
// be deleted.
|
||||||
|
func TestPruneBlobs_FailedDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
|
||||||
|
// The store holds no manifest, so nothing references this blob.
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
|
||||||
|
store := faultstore.New(env.store)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
env.v.Storage = store
|
||||||
|
|
||||||
|
err := env.v.PruneBlobs(&vaultik.PruneOptions{Force: true})
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPruneBlobs_JSONFailedDeleteReturnsError checks that prune --json
|
||||||
|
// prints its document and then returns an error when an unreferenced
|
||||||
|
// blob cannot be deleted.
|
||||||
|
func TestPruneBlobs_JSONFailedDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
|
||||||
|
store := faultstore.New(env.store)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
env.v.Storage = store
|
||||||
|
|
||||||
|
err := env.v.PruneBlobs(&vaultik.PruneOptions{JSON: true})
|
||||||
|
|
||||||
|
require.Error(t, err, "prune must fail when a blob cannot be deleted")
|
||||||
|
|
||||||
|
var result vaultik.PruneBlobsResult
|
||||||
|
|
||||||
|
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &result),
|
||||||
|
"stdout must hold the JSON document, got:\n%s", env.stdout.String())
|
||||||
|
assert.Equal(t, 1, result.BlobsFailed)
|
||||||
|
assert.Equal(t, 0, result.BlobsDeleted)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestCreateSnapshotPrune_FailedBlobDeleteReturnsError checks that the
|
||||||
|
// prune step of snapshot create --prune fails the run when an
|
||||||
|
// unreferenced blob cannot be deleted.
|
||||||
|
func TestCreateSnapshotPrune_FailedBlobDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
// No snapshot references this blob, so the prune step deletes it.
|
||||||
|
orphan := filepath.Join(storeDir, "blobs", testBlobHashA[:2],
|
||||||
|
testBlobHashA[2:4], testBlobHashA)
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Dir(orphan), 0o750))
|
||||||
|
require.NoError(t, os.WriteFile(orphan, []byte("orphan"), 0o600))
|
||||||
|
|
||||||
|
store := faultstore.New(v.Storage)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
v.Storage = store
|
||||||
|
|
||||||
|
err := v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
Cron: true,
|
||||||
|
Prune: true,
|
||||||
|
})
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||||
|
assert.FileExists(t, orphan)
|
||||||
|
}
|
||||||
@@ -476,7 +476,8 @@ func TestRemoveAllSnapshots_WithForce(t *testing.T) {
|
|||||||
out := tv.Stdout.String()
|
out := tv.Stdout.String()
|
||||||
assert.Contains(t, out, "Removed 2 snapshot(s)")
|
assert.Contains(t, out, "Removed 2 snapshot(s)")
|
||||||
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
||||||
assert.Contains(t, out, "vaultik prune")
|
// Remote nuke, the only caller, deletes the blobs next itself.
|
||||||
|
assert.NotContains(t, out, "vaultik prune")
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRemoveAllSnapshots_DryRun(t *testing.T) {
|
func TestRemoveAllSnapshots_DryRun(t *testing.T) {
|
||||||
|
|||||||
+27
-10
@@ -103,11 +103,12 @@ type RestoreResult struct {
|
|||||||
BlobsDownloaded int
|
BlobsDownloaded int
|
||||||
BytesDownloaded int64
|
BytesDownloaded int64
|
||||||
Duration time.Duration
|
Duration time.Duration
|
||||||
|
FilesFailed int
|
||||||
|
FailedFiles []string // Paths of files that failed to restore
|
||||||
// Verification results (only populated if Verify option is set)
|
// Verification results (only populated if Verify option is set)
|
||||||
FilesVerified int
|
FilesVerified int
|
||||||
BytesVerified int64
|
BytesVerified int64
|
||||||
FilesFailed int
|
VerifyFailedFiles []string // Paths of restored files that failed verification
|
||||||
FailedFiles []string // Paths of files that failed verification
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Restore restores files from a snapshot to the target directory
|
// Restore restores files from a snapshot to the target directory
|
||||||
@@ -658,20 +659,34 @@ func (v *Vaultik) handleRestoreVerification(
|
|||||||
opts *RestoreOptions,
|
opts *RestoreOptions,
|
||||||
result *RestoreResult,
|
result *RestoreResult,
|
||||||
) error {
|
) error {
|
||||||
err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result)
|
// A file that failed to restore was not written and is already
|
||||||
|
// reported as such, so it is not verified.
|
||||||
|
failedToRestore := make(map[string]bool, len(result.FailedFiles))
|
||||||
|
for _, path := range result.FailedFiles {
|
||||||
|
failedToRestore[path] = true
|
||||||
|
}
|
||||||
|
|
||||||
|
restored := make([]*database.File, 0, len(files))
|
||||||
|
for _, file := range files {
|
||||||
|
if !failedToRestore[file.Path.String()] {
|
||||||
|
restored = append(restored, file)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
err := v.verifyRestoredFiles(v.ctx, repos, restored, opts.TargetDir, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("verification failed: %w", err)
|
return fmt.Errorf("verification failed: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if result.FilesFailed > 0 {
|
if len(result.VerifyFailedFiles) > 0 {
|
||||||
v.UI.Errorf("Verification failed: %s files did not match expected checksums.",
|
v.UI.Errorf("Verification failed: %s files did not match expected checksums.",
|
||||||
v.UI.Count(result.FilesFailed))
|
v.UI.Count(len(result.VerifyFailedFiles)))
|
||||||
|
|
||||||
for _, path := range result.FailedFiles {
|
for _, path := range result.VerifyFailedFiles {
|
||||||
v.UI.Detailf("%s", v.UI.Path(path))
|
v.UI.Detailf("%s", v.UI.Path(path))
|
||||||
}
|
}
|
||||||
|
|
||||||
return fmt.Errorf("%d %w", result.FilesFailed, errFilesFailedVerify)
|
return fmt.Errorf("%d %w", len(result.VerifyFailedFiles), errFilesFailedVerify)
|
||||||
}
|
}
|
||||||
|
|
||||||
v.UI.Completef("Verified %s files (%s).",
|
v.UI.Completef("Verified %s files (%s).",
|
||||||
@@ -702,6 +717,8 @@ func (v *Vaultik) downloadSnapshotDB(
|
|||||||
return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err)
|
return nil, "", fmt.Errorf("downloading %s: %w", dbKey, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
reader = newCloseOnCancelReader(v.ctx, reader)
|
||||||
|
|
||||||
defer func() { _ = reader.Close() }()
|
defer func() { _ = reader.Close() }()
|
||||||
|
|
||||||
// Decrypt and decompress straight from the storage stream, then stream
|
// Decrypt and decompress straight from the storage stream, then stream
|
||||||
@@ -1431,8 +1448,8 @@ func (v *Vaultik) verifyRestoredFiles(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("File verification failed", "path", file.Path, "error", err)
|
log.Error("File verification failed", "path", file.Path, "error", err)
|
||||||
|
|
||||||
result.FilesFailed++
|
result.VerifyFailedFiles = append(
|
||||||
result.FailedFiles = append(result.FailedFiles, file.Path.String())
|
result.VerifyFailedFiles, file.Path.String())
|
||||||
}
|
}
|
||||||
|
|
||||||
bytesProcessed += file.Size
|
bytesProcessed += file.Size
|
||||||
@@ -1448,7 +1465,7 @@ func (v *Vaultik) verifyRestoredFiles(
|
|||||||
log.Info("Verification complete",
|
log.Info("Verification complete",
|
||||||
"files_verified", result.FilesVerified,
|
"files_verified", result.FilesVerified,
|
||||||
"bytes_verified", ubytes(result.BytesVerified),
|
"bytes_verified", ubytes(result.BytesVerified),
|
||||||
"files_failed", result.FilesFailed,
|
"files_failed", len(result.VerifyFailedFiles),
|
||||||
)
|
)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -207,30 +207,37 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
|||||||
"the cancel was reported as a failed file")
|
"the cancel was reported as a failed file")
|
||||||
}
|
}
|
||||||
|
|
||||||
// stallingBlobStorer wraps a Storer so that every blob download hands
|
// stallingStorer wraps a Storer so that each download of a key chosen by
|
||||||
// over the first half of the blob and then stalls, ignoring the context,
|
// stalls hands over the first half of the object and then stalls,
|
||||||
// as a file:// blob on a hung USB stick or network share does. Only
|
// ignoring the context, as a file:// read on a hung USB stick or network
|
||||||
// closing the returned reader ends the stall.
|
// share does. Only closing the returned reader ends the stall.
|
||||||
type stallingBlobStorer struct {
|
type stallingStorer struct {
|
||||||
storage.Storer
|
storage.Storer
|
||||||
|
|
||||||
|
stalls func(key string) bool
|
||||||
once sync.Once
|
once sync.Once
|
||||||
stalled chan struct{}
|
stalled chan struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
|
func newStallingStorer(
|
||||||
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
|
inner storage.Storer, stalls func(key string) bool,
|
||||||
|
) *stallingStorer {
|
||||||
|
return &stallingStorer{
|
||||||
|
Storer: inner,
|
||||||
|
stalls: stalls,
|
||||||
|
stalled: make(chan struct{}),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *stallingBlobStorer) Get(
|
func (s *stallingStorer) Get(
|
||||||
ctx context.Context, key string,
|
ctx context.Context, key string,
|
||||||
) (io.ReadCloser, error) {
|
) (io.ReadCloser, error) {
|
||||||
rc, err := s.Storer.Get(ctx, key)
|
rc, err := s.Storer.Get(ctx, key)
|
||||||
if err != nil || !strings.HasPrefix(key, "blobs/") {
|
if err != nil || !s.stalls(key) {
|
||||||
return rc, err
|
return rc, err
|
||||||
}
|
}
|
||||||
|
|
||||||
blob, err := io.ReadAll(rc)
|
object, err := io.ReadAll(rc)
|
||||||
_ = rc.Close()
|
_ = rc.Close()
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -242,7 +249,7 @@ func (s *stallingBlobStorer) Get(
|
|||||||
pr, pw := io.Pipe()
|
pr, pw := io.Pipe()
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
_, _ = pw.Write(blob[:len(blob)/2])
|
_, _ = pw.Write(object[:len(object)/2])
|
||||||
|
|
||||||
s.once.Do(func() { close(s.stalled) })
|
s.once.Do(func() { close(s.stalled) })
|
||||||
}()
|
}()
|
||||||
@@ -250,17 +257,26 @@ func (s *stallingBlobStorer) Get(
|
|||||||
return pr, nil
|
return pr, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
|
// TestInterruptDuringStalledReadCleansTempDir stops a restore and a deep
|
||||||
// deep verify while a blob read is stalled on a source that ignores the
|
// verify while a read of a blob or of the snapshot database is stalled
|
||||||
// context. Stopping must end that read, so the operation unwinds and
|
// on a source that ignores the context. Stopping must end that read, so
|
||||||
// removes its decrypted scratch files well within the shutdown timeout
|
// the operation unwinds and removes its decrypted scratch files well
|
||||||
// (issue #325).
|
// within the shutdown timeout (issues #325 and #356).
|
||||||
//
|
//
|
||||||
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
|
||||||
// so it can assert on exactly the scratch files the operation created.
|
// so it can assert on exactly the scratch files the operation created.
|
||||||
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
func TestInterruptDuringStalledReadCleansTempDir(t *testing.T) {
|
||||||
log.Initialize(log.Config{})
|
log.Initialize(log.Config{})
|
||||||
|
|
||||||
|
stalledReads := map[string]func(key string) bool{
|
||||||
|
"blob": func(key string) bool {
|
||||||
|
return strings.HasPrefix(key, "blobs/")
|
||||||
|
},
|
||||||
|
"snapshot database": func(key string) bool {
|
||||||
|
return strings.HasSuffix(key, "/db.zst.age")
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
|
||||||
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
|
||||||
return v.Restore(&RestoreOptions{
|
return v.Restore(&RestoreOptions{
|
||||||
@@ -273,8 +289,9 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
for name, op := range operations {
|
for readName, stalls := range stalledReads {
|
||||||
t.Run(name, func(t *testing.T) {
|
for opName, op := range operations {
|
||||||
|
t.Run(readName+" read during "+opName, func(t *testing.T) {
|
||||||
fs := afero.NewOsFs()
|
fs := afero.NewOsFs()
|
||||||
root := t.TempDir()
|
root := t.TempDir()
|
||||||
|
|
||||||
@@ -287,7 +304,7 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
|||||||
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
require.NoError(t, fs.MkdirAll(scratch, 0o755))
|
||||||
t.Setenv("TMPDIR", scratch)
|
t.Setenv("TMPDIR", scratch)
|
||||||
|
|
||||||
stalling := newStallingBlobStorer(storer)
|
stalling := newStallingStorer(storer, stalls)
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -304,10 +321,14 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
|||||||
select {
|
select {
|
||||||
case <-stalling.stalled:
|
case <-stalling.stalled:
|
||||||
case <-time.After(30 * time.Second):
|
case <-time.After(30 * time.Second):
|
||||||
t.Fatal("the operation never reached a blob read")
|
t.Fatal("the operation never reached the stalled read")
|
||||||
}
|
}
|
||||||
|
|
||||||
require.NotEmpty(t, scratchEntries(t, scratch),
|
// The snapshot database's temp directory is created just
|
||||||
|
// after its read starts, so it can appear after the stall.
|
||||||
|
require.Eventually(t, func() bool {
|
||||||
|
return len(scratchEntries(t, scratch)) > 0
|
||||||
|
}, 10*time.Second, 10*time.Millisecond,
|
||||||
"expected decrypted scratch files to exist mid-read")
|
"expected decrypted scratch files to exist mid-read")
|
||||||
|
|
||||||
stopCtx, stopCancel := context.WithTimeout(
|
stopCtx, stopCancel := context.WithTimeout(
|
||||||
@@ -315,10 +336,11 @@ func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
|
|||||||
defer stopCancel()
|
defer stopCancel()
|
||||||
|
|
||||||
require.True(t, stop(stopCtx),
|
require.True(t, stop(stopCtx),
|
||||||
"stop timed out; the stalled blob read was not interrupted")
|
"stop timed out; the stalled read was not interrupted")
|
||||||
require.ErrorIs(t, opErr, context.Canceled)
|
require.ErrorIs(t, opErr, context.Canceled)
|
||||||
require.Empty(t, scratchEntries(t, scratch),
|
require.Empty(t, scratchEntries(t, scratch),
|
||||||
"decrypted scratch files remained after the interrupt")
|
"decrypted scratch files remained after the interrupt")
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
@@ -4,7 +4,9 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
@@ -104,6 +106,130 @@ func assertRestoreSkipsLostFiles(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles restores with
|
||||||
|
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||||
|
// deleted. Verification must check only the files that were written, so
|
||||||
|
// a file of the deleted blob is reported as failed to restore and not
|
||||||
|
// also as failing verification, and Restore must still return an error.
|
||||||
|
func TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
backup := backupThenDeleteOneBlob(ctx, t)
|
||||||
|
|
||||||
|
var out bytes.Buffer
|
||||||
|
|
||||||
|
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
|
||||||
|
v.UI = ui.NewWithColor(&out, false)
|
||||||
|
|
||||||
|
err := v.Restore(&vaultik.RestoreOptions{
|
||||||
|
SnapshotID: backup.snapshotID,
|
||||||
|
TargetDir: backup.restoreDir,
|
||||||
|
Verify: true,
|
||||||
|
SkipErrors: true,
|
||||||
|
})
|
||||||
|
|
||||||
|
require.Error(t, err, "restore must fail when files were skipped")
|
||||||
|
assert.Contains(t, err.Error(),
|
||||||
|
fmt.Sprintf("%d file(s) failed to restore", len(backup.lost)))
|
||||||
|
|
||||||
|
written := len(backup.files) - len(backup.lost)
|
||||||
|
|
||||||
|
assert.Contains(t, out.String(), fmt.Sprintf("Verifying %d files", written))
|
||||||
|
assert.Contains(t, out.String(), fmt.Sprintf("Verified %d files", written))
|
||||||
|
assert.NotContains(t, out.String(), "Verification failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRestoreSkipErrorsVerifyReportsFailuresApart restores with
|
||||||
|
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||||
|
// deleted, while one file that does restore is written one byte too long.
|
||||||
|
// The restore warning must list only the files of the deleted blob, the
|
||||||
|
// verification summary only the long file, each path once, and Restore
|
||||||
|
// must return an error.
|
||||||
|
func TestRestoreSkipErrorsVerifyReportsFailuresApart(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
backup := backupThenDeleteOneBlob(ctx, t)
|
||||||
|
|
||||||
|
var long string
|
||||||
|
|
||||||
|
for path := range backup.files {
|
||||||
|
if !backup.lost[path] {
|
||||||
|
long = path
|
||||||
|
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var out bytes.Buffer
|
||||||
|
|
||||||
|
fs := trailingByteFs{Fs: backup.fs, suffix: long}
|
||||||
|
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, fs)
|
||||||
|
v.UI = ui.NewWithColor(&out, false)
|
||||||
|
|
||||||
|
err := v.Restore(&vaultik.RestoreOptions{
|
||||||
|
SnapshotID: backup.snapshotID,
|
||||||
|
TargetDir: backup.restoreDir,
|
||||||
|
Verify: true,
|
||||||
|
SkipErrors: true,
|
||||||
|
})
|
||||||
|
require.Error(t, err, "restore must fail when files failed")
|
||||||
|
|
||||||
|
_, afterWarning, found := strings.Cut(out.String(),
|
||||||
|
fmt.Sprintf("%d file(s) failed to restore:", len(backup.lost)))
|
||||||
|
require.Truef(t, found, "no restore warning in:\n%s", out.String())
|
||||||
|
|
||||||
|
warning, summary, found := strings.Cut(afterWarning,
|
||||||
|
"Verification failed: 1 files did not match")
|
||||||
|
require.Truef(t, found, "no verification summary in:\n%s", out.String())
|
||||||
|
|
||||||
|
for path := range backup.lost {
|
||||||
|
assert.Equalf(t, 1, strings.Count(warning, path),
|
||||||
|
"%s must be listed once as failed to restore", path)
|
||||||
|
assert.NotContainsf(t, summary, path,
|
||||||
|
"%s was not restored and must not fail verification", path)
|
||||||
|
}
|
||||||
|
|
||||||
|
assert.NotContainsf(t, warning, long,
|
||||||
|
"%s was restored and must not be listed as failed to restore", long)
|
||||||
|
assert.Equalf(t, 1, strings.Count(summary, long),
|
||||||
|
"%s must be listed once as failing verification", long)
|
||||||
|
}
|
||||||
|
|
||||||
|
// trailingByteFs appends one byte, on close, to the file it opens whose
|
||||||
|
// path ends in suffix, so that file is restored one byte too long and
|
||||||
|
// fails verification.
|
||||||
|
type trailingByteFs struct {
|
||||||
|
afero.Fs
|
||||||
|
|
||||||
|
suffix string
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:ireturn // afero.Fs.OpenFile is defined to return the interface
|
||||||
|
func (fs trailingByteFs) OpenFile(
|
||||||
|
name string, flag int, perm os.FileMode,
|
||||||
|
) (afero.File, error) {
|
||||||
|
f, err := fs.Fs.OpenFile(name, flag, perm)
|
||||||
|
if err != nil || !strings.HasSuffix(name, fs.suffix) {
|
||||||
|
return f, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return trailingByteFile{File: f}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type trailingByteFile struct {
|
||||||
|
afero.File
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f trailingByteFile) Close() error {
|
||||||
|
_, _ = f.Write([]byte{'x'})
|
||||||
|
|
||||||
|
return f.File.Close()
|
||||||
|
}
|
||||||
|
|
||||||
// TestRestoreMissingBlobAbortsWithoutSkipErrors checks that a deleted blob
|
// TestRestoreMissingBlobAbortsWithoutSkipErrors checks that a deleted blob
|
||||||
// still ends the restore with an error when SkipErrors is not set.
|
// still ends the restore with an error when SkipErrors is not set.
|
||||||
func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
|
func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
|
||||||
|
|||||||
@@ -0,0 +1,83 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"filippo.io/age"
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestVerifyPassesWhenBackupReusesStoredBlob backs up a file, then backs it
|
||||||
|
// up again from an empty local index, as after `database delete`, with an
|
||||||
|
// extra age recipient. The second backup builds the same blob, finds it
|
||||||
|
// already stored and skips the upload, but its own copy is larger, because
|
||||||
|
// age writes a header entry for each recipient. The second snapshot must
|
||||||
|
// record the size of the stored object, so shallow and deep verify pass.
|
||||||
|
func TestVerifyPassesWhenBackupReusesStoredBlob(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
tempDir := t.TempDir()
|
||||||
|
|
||||||
|
dataDir := filepath.Join(tempDir, "source")
|
||||||
|
storeDir := filepath.Join(tempDir, "remote")
|
||||||
|
secondDBPath := filepath.Join(tempDir, "second.sqlite")
|
||||||
|
|
||||||
|
chunkSize := int64(64 * 1024)
|
||||||
|
maxBlobSize := int64(512 * 1024)
|
||||||
|
|
||||||
|
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dataDir, "data.bin"),
|
||||||
|
bytesPattern("reuse-", int(chunkSize*3)), 0o644))
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
runFileStorageBackup(ctx, t, fs, dataDir, storeDir,
|
||||||
|
filepath.Join(tempDir, "first.sqlite"), chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
env := setupDedupBackupEnv(
|
||||||
|
ctx, t, fs, storeDir, secondDBPath, chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
defer func() { _ = env.db.Close() }()
|
||||||
|
|
||||||
|
extra, err := age.GenerateX25519Identity()
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
env.cfg.AgeRecipients = append(env.cfg.AgeRecipients,
|
||||||
|
extra.Recipient().String())
|
||||||
|
|
||||||
|
snapshotID, result := runDedupSnapshot(ctx, t, env.sm, env.makeScanner(),
|
||||||
|
env.cfg.Hostname, "second", dataDir, secondDBPath)
|
||||||
|
require.Positive(t, result.BlobsCreated)
|
||||||
|
require.Zero(t, result.BlobsUploaded,
|
||||||
|
"the second backup should find every blob already stored")
|
||||||
|
|
||||||
|
require.NoError(t, env.db.Close())
|
||||||
|
|
||||||
|
verifier := &vaultik.Vaultik{
|
||||||
|
Config: env.cfg,
|
||||||
|
Storage: env.storer,
|
||||||
|
Fs: fs,
|
||||||
|
Stdout: io.Discard,
|
||||||
|
Stderr: io.Discard,
|
||||||
|
UI: ui.NewWithColor(io.Discard, false),
|
||||||
|
}
|
||||||
|
verifier.SetContext(ctx)
|
||||||
|
|
||||||
|
require.NoError(t,
|
||||||
|
verifier.VerifySnapshotWithOptions(snapshotID, &vaultik.VerifyOptions{}),
|
||||||
|
"shallow verify should pass")
|
||||||
|
require.NoError(t,
|
||||||
|
verifier.VerifySnapshotWithOptions(snapshotID,
|
||||||
|
&vaultik.VerifyOptions{Deep: true}),
|
||||||
|
"deep verify should pass")
|
||||||
|
}
|
||||||
@@ -25,6 +25,7 @@ var (
|
|||||||
errSnapshotVerifyFailed = errors.New("verification failed")
|
errSnapshotVerifyFailed = errors.New("verification failed")
|
||||||
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
||||||
errSnapshotNotFound = errors.New("snapshot not found")
|
errSnapshotNotFound = errors.New("snapshot not found")
|
||||||
|
errSnapshotRemoveFailed = errors.New("snapshot(s) could not be removed")
|
||||||
errInvalidTableName = errors.New("invalid table name")
|
errInvalidTableName = errors.New("invalid table name")
|
||||||
errNoAgeRecipients = errors.New(
|
errNoAgeRecipients = errors.New(
|
||||||
"creating a snapshot needs at least one public key in " +
|
"creating a snapshot needs at least one public key in " +
|
||||||
@@ -242,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(
|
||||||
@@ -282,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) {
|
||||||
@@ -1591,8 +1616,8 @@ func (v *Vaultik) handleRemoveAllDryRun(
|
|||||||
|
|
||||||
// executeRemoveAll deletes every local snapshot and, unless LocalOnly
|
// executeRemoveAll deletes every local snapshot and, unless LocalOnly
|
||||||
// is set, every corresponding remote metadata directory plus any
|
// is set, every corresponding remote metadata directory plus any
|
||||||
// orphan remote keys. Blobs are not touched; the printed prune-command
|
// orphan remote keys. Blobs are not touched. It returns an error when
|
||||||
// hint is the next step.
|
// any snapshot could not be removed.
|
||||||
func (v *Vaultik) executeRemoveAll(
|
func (v *Vaultik) executeRemoveAll(
|
||||||
localSnaps, orphanRemoteKeys []string, opts *RemoveOptions,
|
localSnaps, orphanRemoteKeys []string, opts *RemoveOptions,
|
||||||
) (*RemoveResult, error) {
|
) (*RemoveResult, error) {
|
||||||
@@ -1606,18 +1631,11 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
|
|
||||||
result := &RemoveResult{}
|
result := &RemoveResult{}
|
||||||
|
|
||||||
remoteErrors := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
failures := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
||||||
|
|
||||||
if !opts.LocalOnly {
|
if !opts.LocalOnly {
|
||||||
remoteErrors += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
failures += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
||||||
|
result.RemoteRemoved = failures == 0
|
||||||
if remoteErrors == 0 {
|
|
||||||
result.RemoteRemoved = true
|
|
||||||
} else if v.UI != nil {
|
|
||||||
v.UI.Warningf("Some remote metadata deletions failed. Run '%s' "+
|
|
||||||
"once the remote is healthy to clean up unreferenced blobs.",
|
|
||||||
pruneCommandHint)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if v.SnapshotManager != nil {
|
if v.SnapshotManager != nil {
|
||||||
@@ -1627,6 +1645,10 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if failures > 0 {
|
||||||
|
return result, fmt.Errorf("%d %w", failures, errSnapshotRemoveFailed)
|
||||||
|
}
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return result, v.outputRemoveJSON(result)
|
return result, v.outputRemoveJSON(result)
|
||||||
}
|
}
|
||||||
@@ -1635,9 +1657,6 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
|
|
||||||
if !opts.LocalOnly && result.RemoteRemoved {
|
if !opts.LocalOnly && result.RemoteRemoved {
|
||||||
v.printlnStdout("Removed snapshot metadata from remote storage")
|
v.printlnStdout("Removed snapshot metadata from remote storage")
|
||||||
v.stdoutf("\nNote: Removed snapshots' blobs remain on the remote. "+
|
|
||||||
"Run '%s' to delete any blobs no longer referenced by any "+
|
|
||||||
"remaining remote snapshot.\n", pruneCommandHint)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return result, nil
|
return result, nil
|
||||||
@@ -1680,12 +1699,13 @@ func (v *Vaultik) deleteIncompleteSnapshot(snapshotIDStr string, id any) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// removeAllLocalSnapshots removes each known local snapshot (and, unless
|
// removeAllLocalSnapshots removes each known local snapshot (and, unless
|
||||||
// --local-only, its remote metadata), returning the number of remote
|
// --local-only, its remote metadata), returning the number of snapshots
|
||||||
// deletion failures.
|
// it could not remove. A snapshot left in the local index keeps its
|
||||||
|
// remote metadata too, since it is not an orphan remote key.
|
||||||
func (v *Vaultik) removeAllLocalSnapshots(
|
func (v *Vaultik) removeAllLocalSnapshots(
|
||||||
localSnaps []string, opts *RemoveOptions, result *RemoveResult,
|
localSnaps []string, opts *RemoveOptions, result *RemoveResult,
|
||||||
) int {
|
) int {
|
||||||
remoteErrors := 0
|
failures := 0
|
||||||
|
|
||||||
for _, snapshotID := range localSnaps {
|
for _, snapshotID := range localSnaps {
|
||||||
log.Info("Removing snapshot", "snapshot_id", snapshotID)
|
log.Info("Removing snapshot", "snapshot_id", snapshotID)
|
||||||
@@ -1695,6 +1715,8 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
|||||||
log.Error("Failed to remove from local database",
|
log.Error("Failed to remove from local database",
|
||||||
"snapshot_id", snapshotID, "error", err)
|
"snapshot_id", snapshotID, "error", err)
|
||||||
|
|
||||||
|
failures++
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1705,14 +1727,14 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
|||||||
log.Warn("Failed to remove snapshot metadata from remote",
|
log.Warn("Failed to remove snapshot metadata from remote",
|
||||||
"snapshot_id", snapshotID, "error", err)
|
"snapshot_id", snapshotID, "error", err)
|
||||||
|
|
||||||
remoteErrors++
|
failures++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
result.SnapshotsRemoved = append(result.SnapshotsRemoved, snapshotID)
|
result.SnapshotsRemoved = append(result.SnapshotsRemoved, snapshotID)
|
||||||
}
|
}
|
||||||
|
|
||||||
return remoteErrors
|
return failures
|
||||||
}
|
}
|
||||||
|
|
||||||
// removeAllOrphanRemotes deletes remote snapshot metadata that has no
|
// removeAllOrphanRemotes deletes remote snapshot metadata that has no
|
||||||
|
|||||||
@@ -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))
|
||||||
|
}
|
||||||
@@ -260,6 +260,8 @@ func (v *Vaultik) downloadVerifiedSnapshotDB(
|
|||||||
fmt.Errorf("failed to download database: %w", err))
|
fmt.Errorf("failed to download database: %w", err))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
dbReader = newCloseOnCancelReader(v.ctx, dbReader)
|
||||||
|
|
||||||
defer func() { _ = dbReader.Close() }()
|
defer func() { _ = dbReader.Close() }()
|
||||||
|
|
||||||
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
||||||
|
|||||||
Reference in new issue
Block a user