Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
beefa8c37e |
@@ -375,9 +375,7 @@ 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 restored file's chunk hashes
|
* `--verify`: After restoring, verify every file's chunk hashes match
|
||||||
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. 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
|
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption
|
||||||
- `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 = ?, compressed_size = ?` - Mark blob as uploaded, with the size of the object in storage (the upload is skipped when the blob is already stored)
|
- `UPDATE blobs SET uploaded_ts = ?` - Mark blob as uploaded
|
||||||
- `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
|
||||||
|
|
||||||
|
|||||||
@@ -154,16 +154,13 @@ func (r *BlobRepository) UpdateFinished(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateUploaded marks a blob as uploaded and records the size of the
|
// UpdateUploaded marks a blob as uploaded
|
||||||
// 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, storedSize int64,
|
ctx context.Context, tx *sql.Tx, id string,
|
||||||
) error {
|
) error {
|
||||||
query := `
|
query := `
|
||||||
UPDATE blobs
|
UPDATE blobs
|
||||||
SET uploaded_ts = ?, compressed_size = ?
|
SET uploaded_ts = ?
|
||||||
WHERE id = ?
|
WHERE id = ?
|
||||||
`
|
`
|
||||||
|
|
||||||
@@ -171,9 +168,9 @@ func (r *BlobRepository) UpdateUploaded(
|
|||||||
|
|
||||||
var err error
|
var err error
|
||||||
if tx != nil {
|
if tx != nil {
|
||||||
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
_, err = tx.ExecContext(ctx, query, now, id)
|
||||||
} else {
|
} else {
|
||||||
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
|
_, err = r.db.ExecWithLog(ctx, query, now, 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, with a stored object larger than the local copy
|
// Test UpdateUploaded
|
||||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
||||||
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,10 +141,6 @@ 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) {
|
||||||
|
|||||||
@@ -1517,8 +1517,7 @@ 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(
|
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
||||||
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)
|
||||||
}
|
}
|
||||||
@@ -1554,7 +1553,7 @@ 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)
|
||||||
|
|
||||||
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
blobExists, err := s.uploadBlobIfNeeded(
|
||||||
ctx, blobPath, blobWithReader, startTime, result)
|
ctx, blobPath, blobWithReader, startTime, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||||
@@ -1563,8 +1562,7 @@ func (s *Scanner) handleBlobReady(
|
|||||||
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
|
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = s.recordBlobMetadata(
|
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
||||||
ctx, finishedBlob, storedSize, blobExists, startTime)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
|
||||||
|
|
||||||
@@ -1589,22 +1587,20 @@ 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. It returns the size of the object in storage, which for an
|
// exist, returns whether it existed
|
||||||
// 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,
|
||||||
) (int64, bool, error) {
|
) (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
|
||||||
|
|
||||||
stored, err := s.storage.Stat(ctx, blobPath)
|
_, 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,
|
||||||
@@ -1613,7 +1609,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 stored.Size, true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||||
@@ -1627,7 +1623,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 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
uploadDuration := time.Since(startTime)
|
uploadDuration := time.Since(startTime)
|
||||||
@@ -1658,7 +1654,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||||
}
|
}
|
||||||
|
|
||||||
return finishedBlob.Compressed, false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||||
@@ -1728,7 +1724,6 @@ 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 {
|
||||||
@@ -1740,8 +1735,7 @@ 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(
|
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
|
||||||
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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package vaultik_test
|
package vaultik_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/rand"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -9,7 +11,9 @@ import (
|
|||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
"sneak.berlin/go/vaultik/internal/ui"
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
@@ -77,6 +81,174 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
|
|||||||
"deep verify should fail on a corrupted blob")
|
"deep verify should fail on a corrupted blob")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
|
||||||
|
// that deep verify reports under --json: on a healthy snapshot, with a
|
||||||
|
// damaged blob, and with missing blobs. Deep verify stops at the first
|
||||||
|
// damaged blob, so the blobs read before it are the ones verified; when
|
||||||
|
// blobs are missing it reads no blob contents, but counts every missing
|
||||||
|
// blob.
|
||||||
|
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||||
|
|
||||||
|
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||||
|
|
||||||
|
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, "ok", healthy.Status)
|
||||||
|
require.Len(t, keys, healthy.BlobCount)
|
||||||
|
require.Equal(t, healthy.BlobCount, healthy.Verified)
|
||||||
|
require.Zero(t, healthy.Missing)
|
||||||
|
|
||||||
|
// Damage the last blob read, keeping its length, so every blob before
|
||||||
|
// it is read and passes first.
|
||||||
|
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
|
||||||
|
blob, err := afero.ReadFile(fs, lastPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
blob[len(blob)/2] ^= 0xff
|
||||||
|
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
|
||||||
|
|
||||||
|
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, healthy.BlobCount-1, got.Verified)
|
||||||
|
require.Zero(t, got.Missing)
|
||||||
|
|
||||||
|
// Delete the first blob and the damaged last one: both must be
|
||||||
|
// counted as missing, not only the first one found.
|
||||||
|
var deletedSize int64
|
||||||
|
|
||||||
|
for _, key := range []string{keys[0], keys[len(keys)-1]} {
|
||||||
|
info, statErr := storer.Stat(ctx, key)
|
||||||
|
require.NoError(t, statErr)
|
||||||
|
|
||||||
|
deletedSize += info.Size
|
||||||
|
|
||||||
|
require.NoError(t, storer.Delete(ctx, key))
|
||||||
|
}
|
||||||
|
|
||||||
|
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, 2, got.Missing)
|
||||||
|
require.Equal(t, deletedSize, got.MissingSize)
|
||||||
|
require.Zero(t, got.Verified)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestDeepVerifyJSONCountsWrongSizeBlob checks that under --deep --json a
|
||||||
|
// blob of the wrong size is counted in mismatched and fails the run, and
|
||||||
|
// that the check goes on past it to count a missing blob after it.
|
||||||
|
func TestDeepVerifyJSONCountsWrongSizeBlob(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||||
|
|
||||||
|
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||||
|
|
||||||
|
// Make the first blob checked longer, and delete the last one.
|
||||||
|
firstPath := filepath.Join(storeDir, keys[0])
|
||||||
|
blob, err := afero.ReadFile(fs, firstPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
blob = append(blob, []byte("extra")...)
|
||||||
|
require.NoError(t, afero.WriteFile(fs, firstPath, blob, 0o600))
|
||||||
|
require.NoError(t, storer.Delete(ctx, keys[len(keys)-1]))
|
||||||
|
|
||||||
|
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Equal(t, 1, got.Mismatched)
|
||||||
|
require.Equal(t, 1, got.Missing)
|
||||||
|
require.Zero(t, got.Verified)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestDeepVerifyStopsOnBlobThatCannotBeChecked checks that a blob that is
|
||||||
|
// present in storage but cannot be checked is not counted as missing:
|
||||||
|
// deep verify stops and fails with the storage error.
|
||||||
|
func TestDeepVerifyStopsOnBlobThatCannotBeChecked(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "remote")
|
||||||
|
|
||||||
|
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
|
||||||
|
failing := statFailingStorer{Storer: storer, failKeys: []string{keys[0]}}
|
||||||
|
|
||||||
|
got, err := deepVerifyJSON(ctx, t, cfg, failing, snapshotID)
|
||||||
|
require.Error(t, err)
|
||||||
|
require.Equal(t, "failed", got.Status)
|
||||||
|
require.Zero(t, got.Missing)
|
||||||
|
require.Contains(t, got.ErrorMessage, errStatDenied.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
// backUpSeveralBlobs backs up random data, which does not compress or
|
||||||
|
// deduplicate, so it fills at least three blobs in a store on disk at
|
||||||
|
// storeDir. It returns the blobs' storage keys sorted, which is hash
|
||||||
|
// order, the order deep verify reads them in.
|
||||||
|
func backUpSeveralBlobs(
|
||||||
|
ctx context.Context, t *testing.T, storeDir string,
|
||||||
|
) (*config.Config, *storage.FileStorer, string, []string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
tempDir := t.TempDir()
|
||||||
|
dataDir := filepath.Join(tempDir, "source")
|
||||||
|
dbPath := filepath.Join(tempDir, "index.sqlite")
|
||||||
|
|
||||||
|
chunkSize := int64(32 * 1024)
|
||||||
|
maxBlobSize := int64(128 * 1024)
|
||||||
|
|
||||||
|
data := make([]byte, maxBlobSize*4)
|
||||||
|
_, err := rand.Read(data)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dataDir, "data.bin"), data, 0o644))
|
||||||
|
|
||||||
|
cfg, storer, snapshotID := runFileStorageBackup(
|
||||||
|
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
keys := storedBlobKeys(t, storeDir)
|
||||||
|
require.GreaterOrEqual(t, len(keys), 3)
|
||||||
|
|
||||||
|
return cfg, storer, snapshotID, keys
|
||||||
|
}
|
||||||
|
|
||||||
|
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
|
||||||
|
// the decoded result and the error RunDeepVerify returned.
|
||||||
|
func deepVerifyJSON(
|
||||||
|
ctx context.Context, t *testing.T,
|
||||||
|
cfg *config.Config, storer storage.Storer, snapshotID string,
|
||||||
|
) (vaultik.VerifyResult, error) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var out bytes.Buffer
|
||||||
|
|
||||||
|
v := &vaultik.Vaultik{
|
||||||
|
Config: cfg,
|
||||||
|
Storage: storer,
|
||||||
|
Fs: afero.NewOsFs(),
|
||||||
|
Stdout: &out,
|
||||||
|
Stderr: io.Discard,
|
||||||
|
UI: ui.NewWithColor(io.Discard, false),
|
||||||
|
}
|
||||||
|
v.SetContext(ctx)
|
||||||
|
|
||||||
|
err := v.RunDeepVerify(snapshotID,
|
||||||
|
&vaultik.VerifyOptions{Deep: true, JSON: true})
|
||||||
|
|
||||||
|
return decodeVerifyResult(t, out.Bytes()), err
|
||||||
|
}
|
||||||
|
|
||||||
// corruptOneBlob flips a middle byte of the first blob file found under
|
// corruptOneBlob flips a middle byte of the first blob file found under
|
||||||
// blobsDir, leaving the file length unchanged.
|
// blobsDir, leaving the file length unchanged.
|
||||||
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
|
||||||
|
|||||||
+10
-25
@@ -103,12 +103,11 @@ 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
|
||||||
VerifyFailedFiles []string // Paths of restored files that failed verification
|
FilesFailed int
|
||||||
|
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
|
||||||
@@ -659,34 +658,20 @@ func (v *Vaultik) handleRestoreVerification(
|
|||||||
opts *RestoreOptions,
|
opts *RestoreOptions,
|
||||||
result *RestoreResult,
|
result *RestoreResult,
|
||||||
) error {
|
) error {
|
||||||
// A file that failed to restore was not written and is already
|
err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result)
|
||||||
// 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 len(result.VerifyFailedFiles) > 0 {
|
if result.FilesFailed > 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(len(result.VerifyFailedFiles)))
|
v.UI.Count(result.FilesFailed))
|
||||||
|
|
||||||
for _, path := range result.VerifyFailedFiles {
|
for _, path := range result.FailedFiles {
|
||||||
v.UI.Detailf("%s", v.UI.Path(path))
|
v.UI.Detailf("%s", v.UI.Path(path))
|
||||||
}
|
}
|
||||||
|
|
||||||
return fmt.Errorf("%d %w", len(result.VerifyFailedFiles), errFilesFailedVerify)
|
return fmt.Errorf("%d %w", result.FilesFailed, errFilesFailedVerify)
|
||||||
}
|
}
|
||||||
|
|
||||||
v.UI.Completef("Verified %s files (%s).",
|
v.UI.Completef("Verified %s files (%s).",
|
||||||
@@ -1448,8 +1433,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.VerifyFailedFiles = append(
|
result.FilesFailed++
|
||||||
result.VerifyFailedFiles, file.Path.String())
|
result.FailedFiles = append(result.FailedFiles, file.Path.String())
|
||||||
}
|
}
|
||||||
|
|
||||||
bytesProcessed += file.Size
|
bytesProcessed += file.Size
|
||||||
@@ -1465,7 +1450,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", len(result.VerifyFailedFiles),
|
"files_failed", result.FilesFailed,
|
||||||
)
|
)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -4,9 +4,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/spf13/afero"
|
"github.com/spf13/afero"
|
||||||
@@ -106,130 +104,6 @@ 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) {
|
||||||
|
|||||||
@@ -1,83 +0,0 @@
|
|||||||
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")
|
|
||||||
}
|
|
||||||
+38
-16
@@ -17,6 +17,7 @@ import (
|
|||||||
"sneak.berlin/go/vaultik/internal/database"
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Sentinel errors for snapshot verification failures.
|
// Sentinel errors for snapshot verification failures.
|
||||||
@@ -33,6 +34,8 @@ var (
|
|||||||
errManifestMissingBlob = errors.New(
|
errManifestMissingBlob = errors.New(
|
||||||
"manifest omits blob present in database")
|
"manifest omits blob present in database")
|
||||||
errBlobSizeMismatch = errors.New("blob size mismatch")
|
errBlobSizeMismatch = errors.New("blob size mismatch")
|
||||||
|
errBlobsNotInStorage = errors.New(
|
||||||
|
"blobs missing from storage or stored with the wrong size")
|
||||||
)
|
)
|
||||||
|
|
||||||
// verifyStatusFailed is the JSON status value for a failed verification.
|
// verifyStatusFailed is the JSON status value for a failed verification.
|
||||||
@@ -149,7 +152,6 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
result.Status = "ok"
|
result.Status = "ok"
|
||||||
result.Verified = len(dbBlobs)
|
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputVerifyJSON(result)
|
return v.outputVerifyJSON(result)
|
||||||
@@ -306,7 +308,7 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
v.stdoutf("Checking blob existence in remote storage...\n")
|
v.stdoutf("Checking blob existence in remote storage...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -317,7 +319,8 @@ func (v *Vaultik) runVerificationSteps(
|
|||||||
len(dbBlobs), ubytes(totalSize))
|
len(dbBlobs), ubytes(totalSize))
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
err = v.performDeepVerificationFromDB(
|
||||||
|
dbBlobs, tdb.db.Conn(), opts, result, identities)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
||||||
}
|
}
|
||||||
@@ -669,8 +672,14 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
// verifyBlobExistenceFromDB checks that every blob from the database is
|
||||||
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
// in storage with the size the database records. It checks them all,
|
||||||
|
// counting the missing and wrong-size blobs in result, and fails if it
|
||||||
|
// found any. Any storage error other than not found stops the check and
|
||||||
|
// fails with that error.
|
||||||
|
func (v *Vaultik) verifyBlobExistenceFromDB(
|
||||||
|
blobs []snapshot.BlobInfo, result *VerifyResult,
|
||||||
|
) error {
|
||||||
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
||||||
|
|
||||||
for i, blob := range blobs {
|
for i, blob := range blobs {
|
||||||
@@ -683,17 +692,23 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
// Construct blob path
|
// Construct blob path
|
||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
||||||
|
|
||||||
// Check blob exists
|
|
||||||
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
||||||
if err != nil {
|
switch {
|
||||||
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
case errors.Is(err, storage.ErrNotFound):
|
||||||
}
|
log.Warn("Blob missing from storage", "hash", blob.Hash)
|
||||||
|
|
||||||
// Verify size matches
|
result.Missing++
|
||||||
if stat.Size != blob.CompressedSize {
|
result.MissingSize += blob.CompressedSize
|
||||||
return fmt.Errorf(
|
case err != nil:
|
||||||
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
return fmt.Errorf("checking blob %s in storage: %w",
|
||||||
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
blob.Hash, err)
|
||||||
|
case stat.Size != blob.CompressedSize:
|
||||||
|
log.Warn("Blob has the wrong size in storage",
|
||||||
|
"hash", blob.Hash,
|
||||||
|
"storage_size", stat.Size,
|
||||||
|
"database_size", blob.CompressedSize)
|
||||||
|
|
||||||
|
result.Mismatched++
|
||||||
}
|
}
|
||||||
|
|
||||||
// Progress update every 100 blobs
|
// Progress update every 100 blobs
|
||||||
@@ -707,16 +722,22 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if result.Missing > 0 || result.Mismatched > 0 {
|
||||||
|
return fmt.Errorf("%w: %d missing, %d with the wrong size",
|
||||||
|
errBlobsNotInStorage, result.Missing, result.Mismatched)
|
||||||
|
}
|
||||||
|
|
||||||
log.Info("✓ All blobs exist in storage")
|
log.Info("✓ All blobs exist in storage")
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// performDeepVerificationFromDB downloads and verifies the content of
|
// performDeepVerificationFromDB downloads and verifies the content of
|
||||||
// each blob using the database as source.
|
// each blob using the database as source, stopping at the first blob that
|
||||||
|
// fails. Each blob that passes is counted in result.Verified.
|
||||||
func (v *Vaultik) performDeepVerificationFromDB(
|
func (v *Vaultik) performDeepVerificationFromDB(
|
||||||
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
||||||
identities []age.Identity,
|
result *VerifyResult, identities []age.Identity,
|
||||||
) error {
|
) error {
|
||||||
// Calculate total bytes for ETA
|
// Calculate total bytes for ETA
|
||||||
var totalBytesExpected int64
|
var totalBytesExpected int64
|
||||||
@@ -739,6 +760,7 @@ func (v *Vaultik) performDeepVerificationFromDB(
|
|||||||
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
result.Verified++
|
||||||
bytesProcessed += blobInfo.CompressedSize
|
bytesProcessed += blobInfo.CompressedSize
|
||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
remaining := len(blobs) - (i + 1)
|
remaining := len(blobs) - (i + 1)
|
||||||
|
|||||||
Reference in New Issue
Block a user