Compare commits

..
Author SHA1 Message Date
sneak 05acf3dea8 Exit as interrupted when Ctrl-C lands during a local index write (closes #301)
check / check (push) Waiting to run
Once the run's context is cancelled, database/sql rolls an open
transaction back itself, so the rollback in WithTx returned
sql.ErrTxDone. WithTx treated any rollback error as fatal and called
os.Exit(1): an interrupted backup printed "FATAL: failed to rollback
transaction", exited 1 instead of 130 and skipped its deferred cleanup.

WithTx now ignores sql.ErrTxDone from the rollback and returns the
body's error. A rollback that really fails is returned together with
the body's error instead of ending the process; on a panic it is
dropped and the panic carries on. The tests make the rollback really
fail by running ROLLBACK inside the body.

Model: opus-5-5
2026-10-13 08:25:42 +00:00
19 changed files with 162 additions and 456 deletions
+1 -4
View File
@@ -433,10 +433,7 @@ object store. Credentials are read from `s3.access_key_id` and
`s3.secret_access_key` in the config file.
**Local filesystem** (`file:///path/to/backup`): Stores blobs and metadata on
a local or mounted filesystem. Useful for testing or backing up to a NAS. The
path must be absolute; a relative one is refused when the config loads.
Percent-escapes in the path are decoded, so `my%20stick` and `my stick` name
the same directory.
a local or mounted filesystem. Useful for testing or backing up to a NAS.
**Rclone** (`rclone://remote/path`): Uses rclone's 70+ supported cloud
providers. Requires rclone to be configured separately (`rclone config`).
-11
View File
@@ -245,17 +245,6 @@ func (p *Packer) Flush(ctx context.Context) error {
return nil
}
// DiscardCurrentBlob closes and removes the temp file of the blob being
// packed, if any, and drops that blob. A caller that stops before Flush
// must call it, or the temp file is left behind. Thread-safe.
func (p *Packer) DiscardCurrentBlob() {
p.mu.Lock()
defer p.mu.Unlock()
p.cleanupTempFile()
p.currentBlob = nil
}
// FinalizeBlob finalizes the current blob being assembled.
// This compresses the accumulated chunks, encrypts the result, and computes
// the content-addressed hash. The finalized blob is either passed to the
-6
View File
@@ -251,12 +251,6 @@ func NewConfigCommand() *cobra.Command {
Use: "config",
Short: "Manage the configuration file",
Long: "Commands for creating, editing, and querying the vaultik config file.",
// cobra checks Args only on a command that has a Run: without
// one, a mistyped subcommand printed help and exited 0.
Args: cobra.NoArgs,
Run: func(cmd *cobra.Command, _ []string) {
_ = cmd.Help()
},
}
cmd.AddCommand(newConfigInitCommand())
-6
View File
@@ -15,12 +15,6 @@ func NewDatabaseCommand() *cobra.Command {
Use: "database",
Short: "Manage the local state database",
Long: `Commands for managing the local SQLite state database.`,
// cobra checks Args only on a command that has a Run: without
// one, a mistyped subcommand printed help and exited 0.
Args: cobra.NoArgs,
Run: func(cmd *cobra.Command, _ []string) {
_ = cmd.Help()
},
}
cmd.AddCommand(
@@ -1,43 +0,0 @@
package cli //nolint:testpackage // shares runEntry and the command-word constants
import (
"testing"
"github.com/stretchr/testify/assert"
)
// TestEntryGroupRejectsUnknownSubcommand runs each group command with a
// mistyped subcommand, as a crontab line might carry one. It must fail
// with exit 1 and name the word, as `vaultik bogus` does at the top
// level, instead of printing the group's help and exiting 0 having done
// nothing. The group name alone still prints its help.
//
//nolint:paralleltest // replaces os.Args, os.Stdout and os.Stderr
func TestEntryGroupRejectsUnknownSubcommand(t *testing.T) {
for _, testCase := range []struct {
group string
mistyped string
}{
{group: cmdSnapshot, mistyped: "craete"},
{group: "config", mistyped: "inti"},
{group: cmdRemote, mistyped: "ifno"},
{group: "database", mistyped: "delte"},
{group: "completion", mistyped: "nosuchshell"},
} {
t.Run(testCase.group, func(t *testing.T) {
code, stdout, stderr := runEntry(t,
testCase.group, testCase.mistyped)
assert.Equal(t, 1, code)
assert.Empty(t, stdout)
assert.Contains(t, stderr,
`unknown command "`+testCase.mistyped+`"`)
code, stdout, _ = runEntry(t, testCase.group)
assert.Equal(t, 0, code)
assert.Contains(t, stdout,
programName+" "+testCase.group+" [command]")
})
}
}
-6
View File
@@ -18,12 +18,6 @@ func NewRemoteCommand() *cobra.Command {
Use: "remote",
Short: "Remote storage management commands",
Long: "Commands for inspecting and managing remote storage",
// cobra checks Args only on a command that has a Run: without
// one, a mistyped subcommand printed help and exited 0.
Args: cobra.NoArgs,
Run: func(cmd *cobra.Command, _ []string) {
_ = cmd.Help()
},
}
// Add subcommands
-15
View File
@@ -74,21 +74,6 @@ on the source system.`,
NewDatabaseCommand(),
)
// cobra adds the completion command only when Execute runs, and with
// no Run, so it skipped its own cobra.NoArgs and a mistyped shell name
// printed help and exited 0. Adding it here lets it have a Run. Its
// scripts go to os.Stdout as it is now: a later SetOut does not
// redirect them.
cmd.InitDefaultCompletionCmd()
for _, sub := range cmd.Commands() {
if sub.Name() == "completion" {
sub.Run = func(completion *cobra.Command, _ []string) {
_ = completion.Help()
}
}
}
return cmd
}
-6
View File
@@ -41,12 +41,6 @@ func NewSnapshotCommand() *cobra.Command {
Use: "snapshot",
Short: "Snapshot management commands",
Long: "Commands for creating, listing, and managing snapshots",
// cobra checks Args only on a command that has a Run: without
// one, a mistyped subcommand printed help and exited 0.
Args: cobra.NoArgs,
Run: func(cmd *cobra.Command, _ []string) {
_ = cmd.Help()
},
}
// Add subcommands
+2 -12
View File
@@ -61,9 +61,6 @@ var (
errBadS3PartSize = errors.New("s3.part_size must be between 5MiB and 5GiB")
errBadStorageScheme = errors.New(
"storage_url must start with s3://, file://, or rclone://")
errFileURLNotAbsolute = errors.New(
"a file:// storage_url needs an absolute path, " +
"as in file:///path/to/backup")
errStorageNotConfigured = errors.New(
"storage not configured; set storage_url or provide s3.endpoint + " +
"s3.bucket + credentials")
@@ -464,18 +461,11 @@ func (c *Config) validateStorage() error {
// validateStorageURL validates URL-based storage configuration. File and
// rclone URLs need no credentials; S3 URLs require the legacy s3.*
// credential fields. A file URL must name an absolute path: a relative one
// would resolve against the working directory, so the destination would
// move with it.
// credential fields.
func (c *Config) validateStorageURL() error {
switch {
case strings.HasPrefix(c.StorageURL, "file://"):
// storage.ParseStorageURL applies the same rule, but config
// cannot import storage, which imports config.
if !strings.HasPrefix(c.StorageURL, "file:///") {
return errFileURLNotAbsolute
}
// File storage doesn't need S3 credentials
return nil
case strings.HasPrefix(c.StorageURL, "rclone://"):
// Rclone storage uses rclone's own config
-29
View File
@@ -337,35 +337,6 @@ func TestLoadS3PartSize(t *testing.T) {
})
}
// TestLoadRejectsRelativeFileURL checks that a file:// storage_url with a
// relative path fails at load. Accepted, it would resolve against the
// working directory, so a backup run from another directory would write
// to a different destination and still succeed.
func TestLoadRejectsRelativeFileURL(t *testing.T) {
t.Parallel()
const relativeConfig = "snapshots:\n" +
" test:\n" +
" paths: [/tmp/vaultik-test-source]\n" +
"storage_url: file://dest\n"
path := filepath.Join(t.TempDir(), "config.yml")
err := os.WriteFile(path, []byte(relativeConfig), 0o600)
if err != nil {
t.Fatalf("write config: %v", err)
}
_, err = Load(path)
if !errors.Is(err, errFileURLNotAbsolute) {
t.Fatalf("Load() error = %v, want errFileURLNotAbsolute", err)
}
if !strings.Contains(err.Error(), "absolute path") {
t.Errorf("error %q does not say to use an absolute path", err.Error())
}
}
// TestValidateAgeRecipients checks that recipients are parsed at config load
// (a bad entry fails immediately, not mid-backup) and that no invalid entry —
// least of all a pasted secret key — is echoed in the error. An empty list
+12 -9
View File
@@ -3,6 +3,7 @@ package database
import (
"context"
"database/sql"
"errors"
"fmt"
)
@@ -60,22 +61,24 @@ func (r *Repositories) WithTx(ctx context.Context, fn TxFunc) error {
defer func() {
if p := recover(); p != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
Fatalf("failed to rollback transaction: %v", rollbackErr)
}
// The panic is the failure to report, so a rollback error
// is dropped.
_ = tx.Rollback()
panic(p)
} else if err != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
Fatalf("failed to rollback transaction: %v", rollbackErr)
}
}
}()
err = fn(ctx, tx)
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
}
+116
View File
@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"errors"
"strings"
"testing"
"time"
@@ -235,6 +236,121 @@ func TestRepositoriesTransactionRollback(t *testing.T) {
}
}
// When the context is cancelled during the body, database/sql rolls the
// transaction back itself, so WithTx's own rollback finds it already
// done. WithTx must still return the body's error.
func TestRepositoriesTransactionContextCancelled(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.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
file := &database.File{
Path: "/test/cancelled_file.txt",
MTime: time.Now().Truncate(time.Second),
Size: 1024,
Mode: 0644,
UID: 1000,
GID: 1000,
}
err := repos.Files.Create(ctx, tx, file)
if err != nil {
return err
}
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)
}
file, err := repos.Files.GetByPath(
context.Background(), "/test/cancelled_file.txt",
)
if err != nil {
t.Fatalf("error checking for file: %v", err)
}
if file != nil {
t.Error("file should not exist after the cancelled transaction")
}
}
// A body that ends the transaction itself with ROLLBACK makes WithTx's
// own rollback fail. WithTx must return that failure together with the
// body's error.
func TestRepositoriesTransactionRollbackFails(t *testing.T) {
t.Parallel()
db, cleanup := setupTestDB(t)
defer cleanup()
ctx := context.Background()
repos := database.NewRepositories(db)
err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "ROLLBACK")
if err != nil {
return err
}
return errIntentionalRollback
})
if !errors.Is(err, errIntentionalRollback) {
t.Fatalf("expected the body's error, got: %v", err)
}
if !strings.Contains(err.Error(), "rolling back transaction") {
t.Errorf("expected the failed rollback in the error, got: %v", err)
}
}
// When the body ends the transaction itself with ROLLBACK and then
// panics, WithTx's own rollback fails; the panic must still reach the
// caller.
func TestRepositoriesTransactionPanicRollbackFails(t *testing.T) {
t.Parallel()
db, cleanup := setupTestDB(t)
defer cleanup()
ctx := context.Background()
repos := database.NewRepositories(db)
defer func() {
if recover() == nil {
t.Error("expected the body's panic to reach the caller")
}
}()
_ = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "ROLLBACK")
if err != nil {
return err
}
panic(errIntentionalRollback)
})
}
func TestRepositoriesReadTransaction(t *testing.T) {
t.Parallel()
-47
View File
@@ -531,50 +531,3 @@ func TestExcludePatterns_UnanchoredPatternFile(t *testing.T) {
require.Equal(t, 1, result.FilesScanned,
"Unanchored pattern for file should match anywhere")
}
func TestExcludePatterns_DoubleStarDirectoryAtEveryDepth(t *testing.T) {
t.Parallel()
// As in gitignore, a leading **/ also matches directly under the source dir
fs := setupAnchoredTestFS(t)
scanner, repos, cleanup := createTestScanner(t, fs, []string{"**/projectname"})
defer cleanup()
ctx := context.Background()
createSnapshotRecord(ctx, t, repos, "test-snapshot")
result, err := scanner.Scan(ctx, "/backup", "test-snapshot")
require.NoError(t, err)
// **/projectname should exclude BOTH:
// - /backup/projectname/file.txt
// - /backup/otherproject/projectname/file.txt
// Total files: 4, excluded: 2
require.Equal(t, 2, result.FilesScanned,
"**/ directory pattern should match at the top level and nested")
}
func TestExcludePatterns_DoubleStarFileAtEveryDepth(t *testing.T) {
t.Parallel()
// As in gitignore, a leading **/ also matches directly under the source dir
fs := setupAnchoredTestFS(t)
scanner, repos, cleanup := createTestScanner(t, fs, []string{"**/*.txt"})
defer cleanup()
ctx := context.Background()
createSnapshotRecord(ctx, t, repos, "test-snapshot")
result, err := scanner.Scan(ctx, "/backup", "test-snapshot")
require.NoError(t, err)
// **/*.txt should exclude ALL .txt files:
// - /backup/file.txt
// - /backup/projectname/file.txt
// - /backup/otherproject/projectname/file.txt
// Total files: 4, excluded: 3
require.Equal(t, 1, result.FilesScanned,
"**/ file pattern should match at the top level and nested")
}
-13
View File
@@ -1295,11 +1295,6 @@ func (s *Scanner) batchAddFilesToSnapshot(
func (s *Scanner) processPhase(
ctx context.Context, filesToProcess []*FileToProcess, result *ScanResult,
) error {
// On an error or cancellation the blob being packed is never
// finalized, so its temp file must be removed here. After a
// successful flush there is no such blob and this does nothing.
defer s.packer.DiscardCurrentBlob()
// Calculate total bytes to process
var totalBytes int64
for _, f := range filesToProcess {
@@ -2018,14 +2013,6 @@ func compileExcludePatterns(patterns []string) []compiledPattern {
// Remove trailing slash if present (directory indicator)
pattern = strings.TrimSuffix(pattern, "/")
// A leading "**/" demands a "/" before the name and so misses an
// entry directly under the source path. An unanchored pattern
// already matches at every depth, so drop the prefix: as in
// gitignore, "**/x" then also matches a top-level "x".
if !anchored {
pattern = strings.TrimPrefix(pattern, "**/")
}
// Compile the glob pattern
// For patterns without path separators, we need to match them as components
// e.g., ".git" should match ".git" anywhere in the path
-57
View File
@@ -381,60 +381,3 @@ func TestScannerCancelStopsSkipErrorsRun(t *testing.T) {
})
}
}
// TestScannerRemovesBlobTempFileWhenRunStops checks that a run stopped
// part-way, by a file it cannot read or by a cancellation, removes the
// temp file of the blob it was packing.
func TestScannerRemovesBlobTempFileWhenRunStops(t *testing.T) {
t.Parallel()
// a.txt is backed up first and starts a blob; the run stops at b.txt.
const target = "/source/b.txt"
tests := []struct {
name string
newFs func(cancel context.CancelFunc) afero.Fs
}{
{
name: "file cannot be read",
newFs: func(context.CancelFunc) afero.Fs {
return &readFailFs{Fs: afero.NewMemMapFs(), target: target}
},
},
{
name: "cancelled",
newFs: func(cancel context.CancelFunc) afero.Fs {
return &cancelOnOpenFs{
Fs: afero.NewMemMapFs(), target: target, cancel: cancel,
}
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
fs := tt.newFs(cancel)
writeSkipErrorTestFile(t, fs, "/source/a.txt", "first file content")
writeSkipErrorTestFile(t, fs, target, "second file content")
_, err := runSkipErrorScan(ctx, t, fs, "/source", false, nil)
if err == nil {
t.Fatal("expected the scan to stop with an error, got nil")
}
left, err := afero.Glob(fs, filepath.Join(os.TempDir(), "vaultik-blob-*"))
if err != nil {
t.Fatalf("listing blob temp files: %v", err)
}
if len(left) != 0 {
t.Fatalf("expected no blob temp file left, got %v", left)
}
})
}
}
+12 -26
View File
@@ -17,10 +17,8 @@ const (
// Sentinel errors for storage URL parsing.
var (
ErrEmptyStorageURL = errors.New("storage URL is empty")
ErrEmptyFilePath = errors.New("file URL path is empty")
ErrRelativeFilePath = errors.New(
"file URL needs an absolute path, as in file:///path/to/backup")
ErrEmptyStorageURL = errors.New("storage URL is empty")
ErrEmptyFilePath = errors.New("file URL path is empty")
ErrMissingBucket = errors.New("s3 URL missing bucket name")
ErrMissingRemote = errors.New("rclone URL missing remote name")
ErrUnsupportedScheme = errors.New(
@@ -53,8 +51,17 @@ func ParseStorageURL(rawURL string) (*URL, error) {
return nil, ErrEmptyStorageURL
}
// Handle file:// URLs
if after, ok := strings.CutPrefix(rawURL, "file://"); ok {
return parseFileURL(after)
path := after
if path == "" {
return nil, ErrEmptyFilePath
}
return &URL{
Scheme: schemeFile,
Prefix: path,
}, nil
}
if strings.HasPrefix(rawURL, "s3://") {
@@ -68,27 +75,6 @@ func ParseStorageURL(rawURL string) (*URL, error) {
return nil, ErrUnsupportedScheme
}
// parseFileURL parses the path that follows file://. The path must be
// absolute: a relative one would resolve against the working directory,
// so the destination would move with it. Percent-escapes are decoded, so
// my%20stick names the directory "my stick"; a literal space is kept.
func parseFileURL(rawPath string) (*URL, error) {
if rawPath == "" {
return nil, ErrEmptyFilePath
}
if !strings.HasPrefix(rawPath, "/") {
return nil, ErrRelativeFilePath
}
path, err := url.PathUnescape(rawPath)
if err != nil {
return nil, fmt.Errorf("invalid file URL path: %w", err)
}
return &URL{Scheme: schemeFile, Prefix: path}, nil
}
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
// the userinfo and any query parameter other than endpoint, region and
// ssl, so a credential-bearing URL is never stored or echoed.
+2 -16
View File
@@ -14,10 +14,7 @@ import (
func TestParseStorageURLValid(t *testing.T) {
t.Parallel()
const (
bucket = "mybucket"
fileScheme = "file"
)
const bucket = "mybucket"
cases := []struct {
name string
@@ -27,17 +24,7 @@ func TestParseStorageURLValid(t *testing.T) {
{
name: "file absolute path",
raw: "file:///var/backups/vaultik",
want: &storage.URL{Scheme: fileScheme, Prefix: "/var/backups/vaultik"},
},
{
name: "file percent-escaped path is decoded",
raw: "file:///Volumes/my%20stick/backup",
want: &storage.URL{Scheme: fileScheme, Prefix: "/Volumes/my stick/backup"},
},
{
name: "file path with a literal space",
raw: "file:///Volumes/my stick/backup",
want: &storage.URL{Scheme: fileScheme, Prefix: "/Volumes/my stick/backup"},
want: &storage.URL{Scheme: "file", Prefix: "/var/backups/vaultik"},
},
{
name: "s3 bucket and prefix, ssl defaults on",
@@ -103,7 +90,6 @@ func TestParseStorageURLErrors(t *testing.T) {
}{
{"empty url", "", storage.ErrEmptyStorageURL},
{"file empty path", "file://", storage.ErrEmptyFilePath},
{"file relative path", "file://dest", storage.ErrRelativeFilePath},
{"s3 missing bucket", "s3://", storage.ErrMissingBucket},
{"s3 missing bucket with path", "s3:///justprefix", storage.ErrMissingBucket},
{"rclone missing remote", "rclone://", storage.ErrMissingRemote},
-115
View File
@@ -1,20 +1,15 @@
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"
)
@@ -82,116 +77,6 @@ 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) {
+17 -35
View File
@@ -32,9 +32,7 @@ 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")
errBlobsNotInStorage = errors.New(
"blobs missing from storage or stored with the wrong size")
errBlobSizeMismatch = errors.New("blob size mismatch")
)
// verifyStatusFailed is the JSON status value for a failed verification.
@@ -137,6 +135,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
}
result.Status = "ok"
result.Verified = len(dbBlobs)
if opts.JSON {
return v.outputVerifyJSON(result)
@@ -291,7 +290,7 @@ func (v *Vaultik) runVerificationSteps(
v.stdoutf("Checking blob existence in remote storage...\n")
}
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
err = v.verifyBlobExistenceFromDB(dbBlobs)
if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err)
}
@@ -302,8 +301,7 @@ func (v *Vaultik) runVerificationSteps(
len(dbBlobs), ubytes(totalSize))
}
err = v.performDeepVerificationFromDB(
dbBlobs, tdb.db.Conn(), opts, result, identities)
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err)
}
@@ -655,13 +653,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
return nil
}
// 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 {
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
for i, blob := range blobs {
@@ -674,21 +667,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
// 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)
switch {
case err != nil:
log.Warn("Blob missing from storage",
"hash", blob.Hash, "error", err)
if err != nil {
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
}
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++
// 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)
}
// Progress update every 100 blobs
@@ -702,22 +691,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
}
}
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, stopping at the first blob that
// fails. Each blob that passes is counted in result.Verified.
// each blob using the database as source.
func (v *Vaultik) performDeepVerificationFromDB(
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
result *VerifyResult, identities []age.Identity,
identities []age.Identity,
) error {
// Calculate total bytes for ETA
var totalBytesExpected int64
@@ -740,7 +723,6 @@ 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)