Compare commits
4
Commits
1ad3faa8ee
...
8585bc87bb
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8585bc87bb | ||
|
|
fb8abcd049 | ||
|
|
6aceaaf086 | ||
|
|
3b2b9cec84 |
@@ -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
|
||||||
|
|||||||
+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
|
||||||
|
|
||||||
|
|||||||
+10
-12
@@ -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))
|
||||||
|
|||||||
@@ -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,7 @@ 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()
|
|
||||||
if err != nil {
|
|
||||||
Fatalf("failed to close rows: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
var files []*File
|
var files []*File
|
||||||
|
|
||||||
@@ -304,12 +294,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 +325,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
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1517,7 +1517,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 +1554,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 +1589,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 +1613,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 +1627,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 +1658,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 +1728,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 +1740,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 +1772,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
|
||||||
|
|||||||
+32
-6
@@ -29,7 +29,10 @@ var (
|
|||||||
ErrURLCredentials = errors.New(
|
ErrURLCredentials = errors.New(
|
||||||
"storage URL must not carry credentials; " +
|
"storage URL must not carry credentials; " +
|
||||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||||
|
ErrRcloneRemoteParams = errors.New(
|
||||||
|
"rclone URL must not carry parameters in the remote name; " +
|
||||||
|
"set them on the remote with rclone config instead")
|
||||||
)
|
)
|
||||||
|
|
||||||
// URL represents a parsed storage URL.
|
// URL represents a parsed storage URL.
|
||||||
@@ -90,8 +93,9 @@ func parseFileURL(rawPath string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||||
// the userinfo and any query parameter other than endpoint, region and
|
// the userinfo or in the endpoint parameter's userinfo, and any query
|
||||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
// parameter other than endpoint, region and ssl, so a credential-bearing
|
||||||
|
// URL is never stored or echoed.
|
||||||
func parseS3URL(rawURL string) (*URL, error) {
|
func parseS3URL(rawURL string) (*URL, error) {
|
||||||
u, err := url.Parse(rawURL)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -114,6 +118,14 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A host name never contains "@", so one in an endpoint, with or
|
||||||
|
// without a scheme, can only introduce userinfo.
|
||||||
|
for _, endpoint := range query["endpoint"] {
|
||||||
|
if strings.Contains(endpoint, "@") {
|
||||||
|
return nil, ErrURLCredentials
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return &URL{
|
return &URL{
|
||||||
Scheme: schemeS3,
|
Scheme: schemeS3,
|
||||||
Bucket: bucket,
|
Bucket: bucket,
|
||||||
@@ -125,9 +137,23 @@ func parseS3URL(rawURL string) (*URL, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||||
// query parameters, so credentials in the userinfo and any parameter at
|
// query parameters, so credentials in the userinfo, parameters in the
|
||||||
// all are rejected rather than silently ignored.
|
// remote name and any query parameter at all are rejected rather than
|
||||||
|
// silently ignored.
|
||||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||||
|
// The remote name runs up to the first "/", "?" or "#". It never
|
||||||
|
// contains a comma; one starts rclone's remote,param=value form,
|
||||||
|
// whose values can be credentials. This is checked before url.Parse,
|
||||||
|
// whose error for such a name can print part of it.
|
||||||
|
remote := strings.TrimPrefix(rawURL, "rclone://")
|
||||||
|
if end := strings.IndexAny(remote, "/?#"); end >= 0 {
|
||||||
|
remote = remote[:end]
|
||||||
|
}
|
||||||
|
|
||||||
|
if strings.Contains(remote, ",") {
|
||||||
|
return nil, ErrRcloneRemoteParams
|
||||||
|
}
|
||||||
|
|
||||||
u, err := url.Parse(rawURL)
|
u, err := url.Parse(rawURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, wrapParseError(err)
|
return nil, wrapParseError(err)
|
||||||
@@ -137,7 +163,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
|||||||
return nil, ErrURLCredentials
|
return nil, ErrURLCredentials
|
||||||
}
|
}
|
||||||
|
|
||||||
remote := u.Host
|
remote = u.Host
|
||||||
if remote == "" {
|
if remote == "" {
|
||||||
return nil, ErrMissingRemote
|
return nil, ErrMissingRemote
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -150,6 +150,19 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
|||||||
wantErr: storage.ErrURLCredentials,
|
wantErr: storage.ErrURLCredentials,
|
||||||
secrets: []string{key, secret, "mybucket"},
|
secrets: []string{key, secret, "mybucket"},
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
name: "s3 userinfo in endpoint",
|
||||||
|
raw: "s3://mybucket?endpoint=https://" +
|
||||||
|
key + ":" + secret + "@minio.example.com",
|
||||||
|
wantErr: storage.ErrURLCredentials,
|
||||||
|
secrets: []string{key, secret},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "s3 userinfo in endpoint without scheme",
|
||||||
|
raw: "s3://mybucket?endpoint=" + key + ":" + secret + "@minio.example.com",
|
||||||
|
wantErr: storage.ErrURLCredentials,
|
||||||
|
secrets: []string{key, secret},
|
||||||
|
},
|
||||||
{
|
{
|
||||||
name: "s3 unknown query param",
|
name: "s3 unknown query param",
|
||||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||||
@@ -203,6 +216,54 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestParseStorageURLRejectsRcloneRemoteParams checks that an rclone
|
||||||
|
// remote name carrying parameters (rclone's "remote,param=value" form,
|
||||||
|
// which can hold a credential) is rejected without echoing the value,
|
||||||
|
// including the URLs that url.Parse itself refuses.
|
||||||
|
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
raw string
|
||||||
|
secrets []string // must not appear in the error message
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "configured remote",
|
||||||
|
raw: "rclone://gdrive,client_secret=topsecret/backups",
|
||||||
|
secrets: []string{"topsecret"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "remote defined inline",
|
||||||
|
raw: "rclone://:s3,access_key_id=KEY,secret_access_key=SECRET/bucket",
|
||||||
|
secrets: []string{"KEY", "SECRET"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "colon in a parameter value",
|
||||||
|
raw: "rclone://gdrive,client_secret=ab:cd/backups",
|
||||||
|
secrets: []string{"cd"},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
_, err := storage.ParseStorageURL(tc.raw)
|
||||||
|
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
|
||||||
|
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||||
|
tc.raw, err, storage.ErrRcloneRemoteParams)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, secret := range tc.secrets {
|
||||||
|
if strings.Contains(err.Error(), secret) {
|
||||||
|
t.Errorf("error message leaked %q: %v", secret, err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse
|
||||||
// itself fails, the wrapped error carries only the inner cause, not the
|
// itself fails, the wrapped error carries only the inner cause, not the
|
||||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||||
|
|||||||
+27
-12
@@ -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).",
|
||||||
@@ -1433,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
|
||||||
@@ -1450,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
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user