diff --git a/internal/vaultik/deep_verify_test.go b/internal/vaultik/deep_verify_test.go index 6c7d782..0592935 100644 --- a/internal/vaultik/deep_verify_test.go +++ b/internal/vaultik/deep_verify_test.go @@ -1,7 +1,9 @@ package vaultik_test import ( + "bytes" "context" + "crypto/rand" "io" "os" "path/filepath" @@ -9,7 +11,9 @@ import ( "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 +81,174 @@ 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() + 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 // 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 3600208..1fd298c 100644 --- a/internal/vaultik/verify.go +++ b/internal/vaultik/verify.go @@ -17,6 +17,7 @@ import ( "sneak.berlin/go/vaultik/internal/database" "sneak.berlin/go/vaultik/internal/log" "sneak.berlin/go/vaultik/internal/snapshot" + "sneak.berlin/go/vaultik/internal/storage" ) // Sentinel errors for snapshot verification failures. @@ -32,7 +33,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. @@ -149,7 +152,6 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error { } result.Status = "ok" - result.Verified = len(dbBlobs) if opts.JSON { return v.outputVerifyJSON(result) @@ -306,7 +308,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) } @@ -317,7 +319,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) } @@ -669,8 +672,14 @@ 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. 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)) for i, blob := range blobs { @@ -683,17 +692,23 @@ 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 errors.Is(err, storage.ErrNotFound): + log.Warn("Blob missing from storage", "hash", blob.Hash) - // 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 err != nil: + return fmt.Errorf("checking blob %s in storage: %w", + 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 @@ -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") 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 @@ -739,6 +760,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)