Compare commits
2
Commits
60c83844ce
...
a92646f7a6
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a92646f7a6 | ||
|
|
09a0c01c43 |
@@ -375,7 +375,9 @@ on the destination in one go, use `vaultik remote nuke --force`.
|
||||
* Optional path arguments to restore specific files/directories (default: all)
|
||||
* Preserves file permissions, timestamps, ownership (ownership requires root),
|
||||
symlinks, and empty directories
|
||||
* `--verify`: After restoring, verify every file's chunk hashes match
|
||||
* `--verify`: After restoring, verify every restored file's chunk hashes
|
||||
match. A file skipped under `--skip-errors` is reported as failed to
|
||||
restore and is not verified.
|
||||
|
||||
**`prune`**: Tidy up everything that isn't needed. Runs three passes:
|
||||
(1) reconcile the local index against the destination store — any
|
||||
|
||||
@@ -103,6 +103,8 @@ func (v *Vaultik) FetchAndDecryptBlob(
|
||||
}
|
||||
|
||||
// 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(
|
||||
ctx context.Context, blobHash string,
|
||||
) (io.ReadCloser, error) {
|
||||
@@ -120,5 +122,47 @@ func (v *Vaultik) FetchBlob(
|
||||
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()
|
||||
}
|
||||
|
||||
+27
-12
@@ -103,11 +103,12 @@ type RestoreResult struct {
|
||||
BlobsDownloaded int
|
||||
BytesDownloaded int64
|
||||
Duration time.Duration
|
||||
FilesFailed int
|
||||
FailedFiles []string // Paths of files that failed to restore
|
||||
// Verification results (only populated if Verify option is set)
|
||||
FilesVerified int
|
||||
BytesVerified int64
|
||||
FilesFailed int
|
||||
FailedFiles []string // Paths of files that failed verification
|
||||
FilesVerified int
|
||||
BytesVerified int64
|
||||
VerifyFailedFiles []string // Paths of restored files that failed verification
|
||||
}
|
||||
|
||||
// Restore restores files from a snapshot to the target directory
|
||||
@@ -647,20 +648,34 @@ func (v *Vaultik) handleRestoreVerification(
|
||||
opts *RestoreOptions,
|
||||
result *RestoreResult,
|
||||
) error {
|
||||
err := v.verifyRestoredFiles(v.ctx, repos, files, opts.TargetDir, result)
|
||||
// A file that failed to restore was not written and is already
|
||||
// reported as such, so it is not verified.
|
||||
failedToRestore := make(map[string]bool, len(result.FailedFiles))
|
||||
for _, path := range result.FailedFiles {
|
||||
failedToRestore[path] = true
|
||||
}
|
||||
|
||||
restored := make([]*database.File, 0, len(files))
|
||||
for _, file := range files {
|
||||
if !failedToRestore[file.Path.String()] {
|
||||
restored = append(restored, file)
|
||||
}
|
||||
}
|
||||
|
||||
err := v.verifyRestoredFiles(v.ctx, repos, restored, opts.TargetDir, result)
|
||||
if err != nil {
|
||||
return fmt.Errorf("verification failed: %w", err)
|
||||
}
|
||||
|
||||
if result.FilesFailed > 0 {
|
||||
if len(result.VerifyFailedFiles) > 0 {
|
||||
v.UI.Errorf("Verification failed: %s files did not match expected checksums.",
|
||||
v.UI.Count(result.FilesFailed))
|
||||
v.UI.Count(len(result.VerifyFailedFiles)))
|
||||
|
||||
for _, path := range result.FailedFiles {
|
||||
for _, path := range result.VerifyFailedFiles {
|
||||
v.UI.Detailf("%s", v.UI.Path(path))
|
||||
}
|
||||
|
||||
return fmt.Errorf("%d %w", result.FilesFailed, errFilesFailedVerify)
|
||||
return fmt.Errorf("%d %w", len(result.VerifyFailedFiles), errFilesFailedVerify)
|
||||
}
|
||||
|
||||
v.UI.Completef("Verified %s files (%s).",
|
||||
@@ -1420,8 +1435,8 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
if err != nil {
|
||||
log.Error("File verification failed", "path", file.Path, "error", err)
|
||||
|
||||
result.FilesFailed++
|
||||
result.FailedFiles = append(result.FailedFiles, file.Path.String())
|
||||
result.VerifyFailedFiles = append(
|
||||
result.VerifyFailedFiles, file.Path.String())
|
||||
}
|
||||
|
||||
bytesProcessed += file.Size
|
||||
@@ -1437,7 +1452,7 @@ func (v *Vaultik) verifyRestoredFiles(
|
||||
log.Info("Verification complete",
|
||||
"files_verified", result.FilesVerified,
|
||||
"bytes_verified", ubytes(result.BytesVerified),
|
||||
"files_failed", result.FilesFailed,
|
||||
"files_failed", len(result.VerifyFailedFiles),
|
||||
)
|
||||
|
||||
return nil
|
||||
|
||||
@@ -142,14 +142,14 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
|
||||
}
|
||||
|
||||
// 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 {
|
||||
t.Helper()
|
||||
|
||||
var matches []string
|
||||
|
||||
for _, pattern := range []string{
|
||||
"vaultik-blobcache-*", "vaultik-restore-*",
|
||||
"vaultik-blobcache-*", "vaultik-restore-*", "vaultik-verify-*",
|
||||
} {
|
||||
found, err := filepath.Glob(filepath.Join(dir, pattern))
|
||||
require.NoError(t, err)
|
||||
@@ -206,3 +206,119 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
|
||||
assert.NotContains(t, out.String(), srcPath,
|
||||
"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")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,9 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
@@ -84,6 +86,130 @@ func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles restores with
|
||||
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||
// deleted. Verification must check only the files that were written, so
|
||||
// a file of the deleted blob is reported as failed to restore and not
|
||||
// also as failing verification, and Restore must still return an error.
|
||||
func TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
backup := backupThenDeleteOneBlob(ctx, t)
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
|
||||
v.UI = ui.NewWithColor(&out, false)
|
||||
|
||||
err := v.Restore(&vaultik.RestoreOptions{
|
||||
SnapshotID: backup.snapshotID,
|
||||
TargetDir: backup.restoreDir,
|
||||
Verify: true,
|
||||
SkipErrors: true,
|
||||
})
|
||||
|
||||
require.Error(t, err, "restore must fail when files were skipped")
|
||||
assert.Contains(t, err.Error(),
|
||||
fmt.Sprintf("%d file(s) failed to restore", len(backup.lost)))
|
||||
|
||||
written := len(backup.files) - len(backup.lost)
|
||||
|
||||
assert.Contains(t, out.String(), fmt.Sprintf("Verifying %d files", written))
|
||||
assert.Contains(t, out.String(), fmt.Sprintf("Verified %d files", written))
|
||||
assert.NotContains(t, out.String(), "Verification failed")
|
||||
}
|
||||
|
||||
// TestRestoreSkipErrorsVerifyReportsFailuresApart restores with
|
||||
// SkipErrors and Verify after one blob of a two-blob snapshot was
|
||||
// deleted, while one file that does restore is written one byte too long.
|
||||
// The restore warning must list only the files of the deleted blob, the
|
||||
// verification summary only the long file, each path once, and Restore
|
||||
// must return an error.
|
||||
func TestRestoreSkipErrorsVerifyReportsFailuresApart(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
backup := backupThenDeleteOneBlob(ctx, t)
|
||||
|
||||
var long string
|
||||
|
||||
for path := range backup.files {
|
||||
if !backup.lost[path] {
|
||||
long = path
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
fs := trailingByteFs{Fs: backup.fs, suffix: long}
|
||||
v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, fs)
|
||||
v.UI = ui.NewWithColor(&out, false)
|
||||
|
||||
err := v.Restore(&vaultik.RestoreOptions{
|
||||
SnapshotID: backup.snapshotID,
|
||||
TargetDir: backup.restoreDir,
|
||||
Verify: true,
|
||||
SkipErrors: true,
|
||||
})
|
||||
require.Error(t, err, "restore must fail when files failed")
|
||||
|
||||
_, afterWarning, found := strings.Cut(out.String(),
|
||||
fmt.Sprintf("%d file(s) failed to restore:", len(backup.lost)))
|
||||
require.Truef(t, found, "no restore warning in:\n%s", out.String())
|
||||
|
||||
warning, summary, found := strings.Cut(afterWarning,
|
||||
"Verification failed: 1 files did not match")
|
||||
require.Truef(t, found, "no verification summary in:\n%s", out.String())
|
||||
|
||||
for path := range backup.lost {
|
||||
assert.Equalf(t, 1, strings.Count(warning, path),
|
||||
"%s must be listed once as failed to restore", path)
|
||||
assert.NotContainsf(t, summary, path,
|
||||
"%s was not restored and must not fail verification", path)
|
||||
}
|
||||
|
||||
assert.NotContainsf(t, warning, long,
|
||||
"%s was restored and must not be listed as failed to restore", long)
|
||||
assert.Equalf(t, 1, strings.Count(summary, long),
|
||||
"%s must be listed once as failing verification", long)
|
||||
}
|
||||
|
||||
// trailingByteFs appends one byte, on close, to the file it opens whose
|
||||
// path ends in suffix, so that file is restored one byte too long and
|
||||
// fails verification.
|
||||
type trailingByteFs struct {
|
||||
afero.Fs
|
||||
|
||||
suffix string
|
||||
}
|
||||
|
||||
//nolint:ireturn // afero.Fs.OpenFile is defined to return the interface
|
||||
func (fs trailingByteFs) OpenFile(
|
||||
name string, flag int, perm os.FileMode,
|
||||
) (afero.File, error) {
|
||||
f, err := fs.Fs.OpenFile(name, flag, perm)
|
||||
if err != nil || !strings.HasSuffix(name, fs.suffix) {
|
||||
return f, err
|
||||
}
|
||||
|
||||
return trailingByteFile{File: f}, nil
|
||||
}
|
||||
|
||||
type trailingByteFile struct {
|
||||
afero.File
|
||||
}
|
||||
|
||||
func (f trailingByteFile) Close() error {
|
||||
_, _ = f.Write([]byte{'x'})
|
||||
|
||||
return f.File.Close()
|
||||
}
|
||||
|
||||
// TestRestoreMissingBlobAbortsWithoutSkipErrors checks that a deleted blob
|
||||
// still ends the restore with an error when SkipErrors is not set.
|
||||
func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user