Compare commits

...
Author SHA1 Message Date
sneak cbe3793aa9 Record the stored size of a blob a backup finds already stored (closes #300)
check / check (push) Waiting to run
When a finished blob already existed at the destination, the upload was
skipped but the blob row, and so the manifest, kept the size of the
local copy. That copy can differ from the stored object when
compression_level or the recipients changed since it was written, and
then shallow and deep verify failed on a good snapshot. Marking a blob
uploaded now records the size of the object in storage: for an existing
blob, the size Stat reports.

Judgement call: the test adds a recipient rather than changing the
compression level, because an extra age recipient always makes the
local copy larger, while two zstd levels need not compress the test
data differently.

Model: opus-5-5
2026-10-13 14:11:29 +00:00
clawbot 09a0c01c43 Stop a stalled blob read when restore or deep verify is interrupted (closes #325)
check / check (push) Waiting to run
Restore and deep verify looked for a cancel only between blobs, and the
file backend's reader ignores the context. A blob read stalled on a
slow USB stick or network share outlasted the 30-second shutdown
timeout, so the process exited before removing the decrypted snapshot
database and blob cache from the temp directory.

FetchBlob, through which both read every blob, now closes the storage
reader as soon as the context is cancelled, ending a blocked read on
any backend; Read then returns the context's error, not the closed
reader's.

Limitation: a read stuck in the kernel that closing cannot interrupt,
such as one on an unresponsive hard NFS mount, still cannot be stopped.

Model: opus-5-5
2026-10-13 15:48:00 +02:00
clawbot 8c54e0ff75 Make a mistyped subcommand under a group command an error (closes #319)
check / check (push) Waiting to run
snapshot, config, remote, database and completion had no Run, and
cobra checks a command's arguments only when it has one, so `vaultik
snapshot craete` printed the group's help and exited 0. Each group now
prints its help only when given no argument; any other word fails as
an unknown command with exit 1, as at the top level.

cobra adds the completion command only when Execute runs, so the root
command now adds it itself to give it a Run. Its scripts then go to
os.Stdout as it was at that moment; a later SetOut does not redirect
them.

Judgement call: the error has no "Did you mean" suggestion, which the
top level gives.

Model: opus-5-5
2026-10-13 14:14:39 +02:00
clawbot d1213ecb34 Make a leading **/ in an exclude pattern match at the top level (closes #303)
check / check (push) Waiting to run
An unanchored exclude pattern such as **/node_modules or **/*.pyc
needed a / before the name, so it excluded sub/node_modules and
sub/y.pyc but backed up node_modules/ and x.pyc directly under the
snapshot path. An unanchored pattern already matches at every depth,
so the leading **/ is now dropped when the pattern is compiled and,
as in gitignore, the pattern matches the top level too.

Judgement call: an anchored /**/x is left as it was; the issue covers
patterns that start with **/.

Model: opus-5-5
2026-10-13 13:50:11 +02:00
14 changed files with 416 additions and 23 deletions
+2 -2
View File
@@ -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 - `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
- `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.
@@ -183,7 +183,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 = ?` - Mark blob as uploaded - `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)
- `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
+6
View File
@@ -251,6 +251,12 @@ func NewConfigCommand() *cobra.Command {
Use: "config", Use: "config",
Short: "Manage the configuration file", Short: "Manage the configuration file",
Long: "Commands for creating, editing, and querying the vaultik config 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()) cmd.AddCommand(newConfigInitCommand())
+6
View File
@@ -15,6 +15,12 @@ func NewDatabaseCommand() *cobra.Command {
Use: "database", Use: "database",
Short: "Manage the local state database", Short: "Manage the local state database",
Long: `Commands for managing the local SQLite 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( cmd.AddCommand(
@@ -0,0 +1,43 @@
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,6 +18,12 @@ func NewRemoteCommand() *cobra.Command {
Use: "remote", Use: "remote",
Short: "Remote storage management commands", Short: "Remote storage management commands",
Long: "Commands for inspecting and managing remote storage", 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 // Add subcommands
+15
View File
@@ -74,6 +74,21 @@ on the source system.`,
NewDatabaseCommand(), 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 return cmd
} }
+6
View File
@@ -41,6 +41,12 @@ func NewSnapshotCommand() *cobra.Command {
Use: "snapshot", Use: "snapshot",
Short: "Snapshot management commands", Short: "Snapshot management commands",
Long: "Commands for creating, listing, and managing snapshots", 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 // Add subcommands
+9 -6
View File
@@ -159,13 +159,16 @@ func (r *BlobRepository) UpdateFinished(
return nil return nil
} }
// UpdateUploaded marks a blob as uploaded // UpdateUploaded marks a blob as uploaded and records the size of the
// 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, ctx context.Context, tx *sql.Tx, id string, storedSize int64,
) error { ) error {
query := ` query := `
UPDATE blobs UPDATE blobs
SET uploaded_ts = ? SET uploaded_ts = ?, compressed_size = ?
WHERE id = ? WHERE id = ?
` `
@@ -173,9 +176,9 @@ func (r *BlobRepository) UpdateUploaded(
var err error var err error
if tx != nil { if tx != nil {
_, err = tx.ExecContext(ctx, query, now, id) _, err = tx.ExecContext(ctx, query, now, storedSize, id)
} else { } else {
_, err = r.db.ExecWithLog(ctx, query, now, id) _, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
} }
if err != nil { if err != nil {
+6 -2
View File
@@ -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 // Test UpdateUploaded, with a stored object larger than the local copy
err = repo.UpdateUploaded(ctx, nil, blob.ID.String()) err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
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,6 +141,10 @@ 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) {
+47
View File
@@ -531,3 +531,50 @@ func TestExcludePatterns_UnanchoredPatternFile(t *testing.T) {
require.Equal(t, 1, result.FilesScanned, require.Equal(t, 1, result.FilesScanned,
"Unanchored pattern for file should match anywhere") "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")
}
+24 -10
View File
@@ -1514,7 +1514,8 @@ 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(ctx, tx, b.ID) err := s.repos.Blobs.UpdateUploaded(
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)
} }
@@ -1550,7 +1551,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)
blobExists, err := s.uploadBlobIfNeeded( storedSize, blobExists, err := s.uploadBlobIfNeeded(
ctx, blobPath, blobWithReader, startTime, result) ctx, blobPath, blobWithReader, startTime, result)
if err != nil { if err != nil {
s.cleanupBlobTempFile(blobWithReader) s.cleanupBlobTempFile(blobWithReader)
@@ -1558,7 +1559,8 @@ func (s *Scanner) handleBlobReady(
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err) return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
} }
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime) err = s.recordBlobMetadata(
ctx, finishedBlob, storedSize, blobExists, startTime)
if err != nil { if err != nil {
s.cleanupBlobTempFile(blobWithReader) s.cleanupBlobTempFile(blobWithReader)
@@ -1580,20 +1582,22 @@ 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, returns whether it existed // exist. It returns the size of the object in storage, which for an
// 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,
) (bool, error) { ) (int64, 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
_, err := s.storage.Stat(ctx, blobPath) stored, 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,
@@ -1602,7 +1606,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 true, nil return stored.Size, true, nil
} }
s.ui.Beginf("Uploading blob %s (%s) to %s.", s.ui.Beginf("Uploading blob %s (%s) to %s.",
@@ -1616,7 +1620,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 false, fmt.Errorf("uploading blob to storage: %w", err) return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
} }
uploadDuration := time.Since(startTime) uploadDuration := time.Since(startTime)
@@ -1647,7 +1651,7 @@ func (s *Scanner) uploadBlobIfNeeded(
stats.BytesUploaded.Add(finishedBlob.Compressed) stats.BytesUploaded.Add(finishedBlob.Compressed)
} }
return false, nil return finishedBlob.Compressed, false, nil
} }
// makeUploadProgressCallback creates a progress callback for blob uploads. // makeUploadProgressCallback creates a progress callback for blob uploads.
@@ -1717,6 +1721,7 @@ 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 {
@@ -1728,7 +1733,8 @@ 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(txCtx, tx, finishedBlob.ID) err := s.repos.Blobs.UpdateUploaded(
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)
} }
@@ -2018,6 +2024,14 @@ func compileExcludePatterns(patterns []string) []compiledPattern {
// Remove trailing slash if present (directory indicator) // Remove trailing slash if present (directory indicator)
pattern = strings.TrimSuffix(pattern, "/") 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 // Compile the glob pattern
// For patterns without path separators, we need to match them as components // For patterns without path separators, we need to match them as components
// e.g., ".git" should match ".git" anywhere in the path // e.g., ".git" should match ".git" anywhere in the path
+45 -1
View File
@@ -103,6 +103,8 @@ func (v *Vaultik) FetchAndDecryptBlob(
} }
// FetchBlob downloads a blob and returns a reader for the encrypted data. // FetchBlob downloads a blob and returns a reader for the encrypted data.
// The reader is closed as soon as ctx is cancelled, which ends a read
// blocked inside it.
func (v *Vaultik) FetchBlob( func (v *Vaultik) FetchBlob(
ctx context.Context, blobHash string, ctx context.Context, blobHash string,
) (io.ReadCloser, error) { ) (io.ReadCloser, error) {
@@ -120,5 +122,47 @@ func (v *Vaultik) FetchBlob(
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err) return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
} }
return rc, nil return newCloseOnCancelReader(ctx, rc), nil
}
// closeOnCancelReader closes the reader it wraps when ctx is cancelled.
// Not every backend's reader stops on cancel by itself (the file
// backend's does not), and a read stalled on a slow or hung source would
// otherwise outlast the shutdown timeout, so the process would exit
// before restore or deep verify removed its decrypted temporary files.
type closeOnCancelReader struct {
io.ReadCloser
ctx context.Context //nolint:containedctx // Read reports its error
stop func() bool
}
func newCloseOnCancelReader(
ctx context.Context, rc io.ReadCloser,
) *closeOnCancelReader {
return &closeOnCancelReader{
ReadCloser: rc,
ctx: ctx,
stop: context.AfterFunc(ctx, func() { _ = rc.Close() }),
}
}
// Read returns the context's error, not the closed reader's, once ctx
// has been cancelled.
func (r *closeOnCancelReader) Read(p []byte) (int, error) {
n, err := r.ReadCloser.Read(p)
if err != nil && r.ctx.Err() != nil {
return n, r.ctx.Err()
}
return n, err
}
// Close closes the wrapped reader, unless the cancel already closed it.
func (r *closeOnCancelReader) Close() error {
if !r.stop() {
return nil
}
return r.ReadCloser.Close()
} }
+118 -2
View File
@@ -142,14 +142,14 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
} }
// scratchEntries returns the vaultik blob-cache and snapshot-database // scratchEntries returns the vaultik blob-cache and snapshot-database
// scratch entries currently present in dir. // scratch entries (restore's and deep verify's) currently present in dir.
func scratchEntries(t *testing.T, dir string) []string { func scratchEntries(t *testing.T, dir string) []string {
t.Helper() t.Helper()
var matches []string var matches []string
for _, pattern := range []string{ for _, pattern := range []string{
"vaultik-blobcache-*", "vaultik-restore-*", "vaultik-blobcache-*", "vaultik-restore-*", "vaultik-verify-*",
} { } {
found, err := filepath.Glob(filepath.Join(dir, pattern)) found, err := filepath.Glob(filepath.Join(dir, pattern))
require.NoError(t, err) require.NoError(t, err)
@@ -206,3 +206,119 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
assert.NotContains(t, out.String(), srcPath, assert.NotContains(t, out.String(), srcPath,
"the cancel was reported as a failed file") "the cancel was reported as a failed file")
} }
// stallingBlobStorer wraps a Storer so that every blob download hands
// over the first half of the blob and then stalls, ignoring the context,
// as a file:// blob on a hung USB stick or network share does. Only
// closing the returned reader ends the stall.
type stallingBlobStorer struct {
storage.Storer
once sync.Once
stalled chan struct{}
}
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
}
func (s *stallingBlobStorer) Get(
ctx context.Context, key string,
) (io.ReadCloser, error) {
rc, err := s.Storer.Get(ctx, key)
if err != nil || !strings.HasPrefix(key, "blobs/") {
return rc, err
}
blob, err := io.ReadAll(rc)
_ = rc.Close()
if err != nil {
return nil, err
}
// The writer is never closed, so once the first half has been read
// the next Read blocks until the reader is closed.
pr, pw := io.Pipe()
go func() {
_, _ = pw.Write(blob[:len(blob)/2])
s.once.Do(func() { close(s.stalled) })
}()
return pr, nil
}
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
// deep verify while a blob read is stalled on a source that ignores the
// context. Stopping must end that read, so the operation unwinds and
// removes its decrypted scratch files well within the shutdown timeout
// (issue #325).
//
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
// so it can assert on exactly the scratch files the operation created.
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
log.Initialize(log.Config{})
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
return v.Restore(&RestoreOptions{
SnapshotID: snapshotID,
TargetDir: targetDir,
})
},
"deep verify": func(v *Vaultik, snapshotID, _ string) error {
return v.RunDeepVerify(snapshotID, &VerifyOptions{Deep: true})
},
}
for name, op := range operations {
t.Run(name, func(t *testing.T) {
fs := afero.NewOsFs()
root := t.TempDir()
// Random content, so the blob stays large enough to stall
// part-way through its compressed, encrypted body.
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
scratch := filepath.Join(root, "scratch")
require.NoError(t, fs.MkdirAll(scratch, 0o755))
t.Setenv("TMPDIR", scratch)
stalling := newStallingBlobStorer(storer)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
v := newRestoreVaultik(ctx, cfg, stalling, fs)
v.cancel = cancel
var opErr error
stop := v.StartOperation(func() {
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
})
select {
case <-stalling.stalled:
case <-time.After(30 * time.Second):
t.Fatal("the operation never reached a blob read")
}
require.NotEmpty(t, scratchEntries(t, scratch),
"expected decrypted scratch files to exist mid-read")
stopCtx, stopCancel := context.WithTimeout(
context.Background(), 10*time.Second)
defer stopCancel()
require.True(t, stop(stopCtx),
"stop timed out; the stalled blob read was not interrupted")
require.ErrorIs(t, opErr, context.Canceled)
require.Empty(t, scratchEntries(t, scratch),
"decrypted scratch files remained after the interrupt")
})
}
}
@@ -0,0 +1,83 @@
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")
}