diff --git a/internal/vaultik/deep_verify_test.go b/internal/vaultik/deep_verify_test.go index 6c7d782..7416d41 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,161 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) { "deep verify should fail on a corrupted blob") } +// TestDeepVerifyJSONCountsBlobs checks the counts that deep verify +// reports under --json: on a healthy snapshot, with a blob that cannot +// be checked, with a damaged blob, with missing blobs, and with a blob +// of the wrong size. Deep verify stops at the first damaged blob, so the +// blobs read before it are the ones verified; when blobs are missing or +// the wrong size it reads no blob contents, but counts every such 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) + + // A blob that is present but cannot be checked is not missing: the + // check stops with the storage error. + denied := &statDeniedStorer{Storer: storer, deniedKey: keys[0]} + + got, err := deepVerifyJSON(ctx, t, cfg, denied, snapshotID) + require.Error(t, err) + require.Equal(t, "failed", got.Status) + require.Zero(t, got.Missing) + require.Contains(t, got.ErrorMessage, os.ErrPermission.Error()) + + // 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) + + // Make a blob between the two deleted ones longer: it is counted as + // the wrong size, and both missing blobs are still counted. + grownPath := filepath.Join(storeDir, keys[1]) + blob, err = afero.ReadFile(fs, grownPath) + require.NoError(t, err) + + blob = append(blob, []byte("extra")...) + require.NoError(t, afero.WriteFile(fs, grownPath, blob, 0o600)) + + 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, 2, got.Missing) + require.Zero(t, got.Verified) +} + +// statDeniedStorer passes every call through to the wrapped Storer, +// except that Stat of deniedKey fails with a permission error, as it +// does for a blob that is present but cannot be read. +type statDeniedStorer struct { + storage.Storer + + deniedKey string +} + +func (s *statDeniedStorer) Stat( + ctx context.Context, key string, +) (*storage.ObjectInfo, error) { + if key == s.deniedKey { + return nil, os.ErrPermission + } + + return s.Storer.Stat(ctx, key) +} + +// 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..6325619 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. @@ -135,7 +138,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 +292,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 +303,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 +656,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 { @@ -667,17 +676,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 @@ -691,16 +706,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 +744,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)