diff --git a/internal/vaultik/deep_verify_test.go b/internal/vaultik/deep_verify_test.go index 6c7d782..c5767bb 100644 --- a/internal/vaultik/deep_verify_test.go +++ b/internal/vaultik/deep_verify_test.go @@ -1,15 +1,20 @@ package vaultik_test import ( + "bytes" "context" + "crypto/rand" "io" "os" "path/filepath" + "slices" "testing" "github.com/spf13/afero" "github.com/stretchr/testify/require" + "sneak.berlin/go/vaultik/internal/config" "sneak.berlin/go/vaultik/internal/log" + "sneak.berlin/go/vaultik/internal/storage" "sneak.berlin/go/vaultik/internal/ui" "sneak.berlin/go/vaultik/internal/vaultik" ) @@ -77,6 +82,116 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) { "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() + tempDir := t.TempDir() + + dataDir := filepath.Join(tempDir, "source") + storeDir := filepath.Join(tempDir, "remote") + dbPath := filepath.Join(tempDir, "index.sqlite") + + chunkSize := int64(32 * 1024) + maxBlobSize := int64(128 * 1024) + + // Random data does not compress or deduplicate, so it fills several + // blobs. + 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)) + + ctx := context.Background() + + cfg, storer, snapshotID := runFileStorageBackup( + ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize) + + healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID) + require.NoError(t, err) + require.Equal(t, "ok", healthy.Status) + require.GreaterOrEqual(t, healthy.BlobCount, 3) + require.Equal(t, healthy.BlobCount, healthy.Verified) + require.Zero(t, healthy.Missing) + + // Deep verify reads blobs in hash order, which is the sorted order of + // their storage keys. + keys, err := storer.List(ctx, "blobs/") + require.NoError(t, err) + slices.Sort(keys) + require.Len(t, keys, healthy.BlobCount) + + // 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) +} + +// 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 // blobsDir, leaving the file length unchanged. func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) { diff --git a/internal/vaultik/verify.go b/internal/vaultik/verify.go index 070c47b..30db0a1 100644 --- a/internal/vaultik/verify.go +++ b/internal/vaultik/verify.go @@ -32,7 +32,9 @@ var ( errManifestExtraBlob = errors.New("manifest contains blob not in database") errManifestMissingBlob = errors.New( "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. @@ -135,7 +137,6 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error { } result.Status = "ok" - result.Verified = len(dbBlobs) if opts.JSON { return v.outputVerifyJSON(result) @@ -290,7 +291,7 @@ func (v *Vaultik) runVerificationSteps( v.stdoutf("Checking blob existence in remote storage...\n") } - err = v.verifyBlobExistenceFromDB(dbBlobs) + err = v.verifyBlobExistenceFromDB(dbBlobs, result) if err != nil { return v.deepVerifyFailure(result, opts, err.Error(), err) } @@ -301,7 +302,8 @@ func (v *Vaultik) runVerificationSteps( 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 { return v.deepVerifyFailure(result, opts, err.Error(), err) } @@ -653,8 +655,13 @@ func (v *Vaultik) verifyManifestAgainstDatabase( return nil } -// verifyBlobExistenceFromDB checks that all blobs from database exist in S3 -func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error { +// verifyBlobExistenceFromDB checks that every blob from the database is +// 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. +func (v *Vaultik) verifyBlobExistenceFromDB( + blobs []snapshot.BlobInfo, result *VerifyResult, +) error { log.Info("Verifying blob existence in S3", "blob_count", len(blobs)) for i, blob := range blobs { @@ -667,17 +674,21 @@ func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error { // Construct blob path 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) - if err != nil { - return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err) - } + switch { + case err != nil: + log.Warn("Blob missing from storage", + "hash", blob.Hash, "error", err) - // Verify size matches - if stat.Size != blob.CompressedSize { - return fmt.Errorf( - "%w: blob %s: S3 has %d bytes, database has %d bytes", - errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize) + result.Missing++ + result.MissingSize += blob.CompressedSize + case stat.Size != blob.CompressedSize: + log.Warn("Blob has the wrong size in storage", + "hash", blob.Hash, + "storage_size", stat.Size, + "database_size", blob.CompressedSize) + + result.Mismatched++ } // Progress update every 100 blobs @@ -691,16 +702,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") return nil } // 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( blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions, - identities []age.Identity, + result *VerifyResult, identities []age.Identity, ) error { // Calculate total bytes for ETA var totalBytesExpected int64 @@ -723,6 +740,7 @@ func (v *Vaultik) performDeepVerificationFromDB( return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err) } + result.Verified++ bytesProcessed += blobInfo.CompressedSize elapsed := time.Since(startTime) remaining := len(blobs) - (i + 1)