Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c0c79cd0a2 | ||
|
|
167c0a0746 | ||
|
|
07c957f84b | ||
|
|
c9a519c233 | ||
|
|
fa56165fe4 | ||
|
|
34f61fb408 | ||
|
|
05bf73c48a | ||
|
|
79a73fa122 |
@@ -349,9 +349,11 @@ the backup destination store. Blobs are NOT touched — deleting blobs
|
|||||||
requires reading every remaining remote manifest (the destination store
|
requires reading every remaining remote manifest (the destination store
|
||||||
may hold snapshots this host doesn't know about), which is what
|
may hold snapshots this host doesn't know about), which is what
|
||||||
`vaultik prune` does. On success the command prints the exact `vaultik
|
`vaultik prune` does. On success the command prints the exact `vaultik
|
||||||
prune` invocation to run as a follow-up. Local row cleanup (files,
|
prune` invocation to run as a follow-up. An ID that names no snapshot in
|
||||||
chunks, blobs the snapshot was the last referrer for) runs
|
the local index or on the destination store is an error; with
|
||||||
automatically. If the destination store is unreachable, the local-DB
|
`--local-only`, the snapshot must be in the local index. Local row
|
||||||
|
cleanup (files, chunks, blobs the snapshot was the last referrer for)
|
||||||
|
runs automatically. If the destination store is unreachable, the local-DB
|
||||||
removal still completes and a warning is emitted; run `vaultik snapshot
|
removal still completes and a warning is emitted; run `vaultik snapshot
|
||||||
remove <snapshot-id>` again once the store is reachable to remove the
|
remove <snapshot-id>` again once the store is reachable to remove the
|
||||||
snapshot's metadata from it (`vaultik prune` does not). To wipe everything
|
snapshot's metadata from it (`vaultik prune` does not). To wipe everything
|
||||||
@@ -382,10 +384,12 @@ manifest on the destination store to compute the still-referenced blob
|
|||||||
set and delete any blob not in that set. Step (3) reads all remote
|
set and delete any blob not in that set. Step (3) reads all remote
|
||||||
manifests — network cost scales with the number of snapshots. `snapshot
|
manifests — network cost scales with the number of snapshots. `snapshot
|
||||||
create --prune` runs the same cleanup automatically; this is the
|
create --prune` runs the same cleanup automatically; this is the
|
||||||
manual entry point for the same work.
|
manual entry point for the same work. If a blob cannot be deleted, the
|
||||||
|
command (and `snapshot create --prune`) exits non-zero.
|
||||||
* `--force`: Skip confirmation prompt
|
* `--force`: Skip confirmation prompt
|
||||||
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
* `--json`: Output stats as JSON. Also skips the confirmation prompt, as
|
||||||
`--force` does.
|
`--force` does. A blob that cannot be deleted is counted in
|
||||||
|
`blobs_failed`, and the document is printed before the non-zero exit.
|
||||||
|
|
||||||
**`info`**: Display system configuration, storage settings, encryption
|
**`info`**: Display system configuration, storage settings, encryption
|
||||||
recipients, and local database statistics.
|
recipients, and local database statistics.
|
||||||
@@ -405,14 +409,16 @@ snapshot's blob count and blob size unknown, `null` in `--json`.
|
|||||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||||
backup destination store, leaving the bucket prefix empty. Destructive and
|
backup destination store, leaving the bucket prefix empty. Destructive and
|
||||||
irreversible. This is the single supported way to wipe the entire
|
irreversible. This is the single supported way to wipe the entire
|
||||||
destination store.
|
destination store. If anything cannot be deleted, the command exits
|
||||||
|
non-zero; running it again deletes what is left.
|
||||||
* `--force`: Required to confirm destruction.
|
* `--force`: Required to confirm destruction.
|
||||||
|
|
||||||
**`database delete`**: Delete the local SQLite state database file
|
**`database delete`**: Delete the local SQLite state database file
|
||||||
entirely. Remote storage is unaffected; the next backup will do a full
|
entirely. Remote storage is unaffected; the next backup will do a full
|
||||||
scan and re-deduplicate against existing remote blobs, and the local
|
scan and re-deduplicate against existing remote blobs, and the local
|
||||||
index will re-bind to the currently configured storage destination.
|
index will re-bind to the currently configured storage destination.
|
||||||
Use this after changing `storage_url` to a different destination.
|
Use this after changing `storage_url`, or `s3.endpoint`, `s3.bucket` or
|
||||||
|
`s3.prefix`, to a different destination.
|
||||||
* `--force`: Skip confirmation prompt
|
* `--force`: Skip confirmation prompt
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
@@ -22,6 +22,70 @@ the tag exists and is exercised; what is left is merging `next` to
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-13: Made `prune` and `remote nuke` exit non-zero when a delete
|
||||||
|
fails ([issue #290](https://git.eeqj.de/sneak/vaultik/issues/290)).
|
||||||
|
`prune`, and the prune step of `snapshot create --prune`, counted a
|
||||||
|
blob it could not delete and still exited 0; it now prints its summary
|
||||||
|
or `--json` document and then fails. `remote nuke` warned about a
|
||||||
|
snapshot's metadata it could not delete, said to run `vaultik prune`,
|
||||||
|
which never removes metadata, and then reported the store empty. It
|
||||||
|
now fails without that report and says to run `vaultik remote nuke
|
||||||
|
--force` again, and a successful nuke no longer advises `prune` for
|
||||||
|
blobs it deletes itself.
|
||||||
|
|
||||||
|
- 2026-10-13: Made `snapshot remove` fail for a snapshot that does not
|
||||||
|
exist ([issue #293](https://git.eeqj.de/sneak/vaultik/issues/293)).
|
||||||
|
The command never checked the ID: it reported the snapshot removed
|
||||||
|
from the local database and from the destination store and exited 0,
|
||||||
|
under `--json`, `--dry-run` and `--local-only` too. An ID that names
|
||||||
|
no snapshot in the local index or on the destination store is now an
|
||||||
|
error; with `--local-only`, the snapshot must be in the local index.
|
||||||
|
The command now reports a removal from either side only when one
|
||||||
|
happened, and `remote_removed` is true only when metadata was deleted.
|
||||||
|
|
||||||
|
- 2026-10-12: Bound the local index to the destination an `s3.*` config
|
||||||
|
names ([issue #298](https://git.eeqj.de/sneak/vaultik/issues/298)).
|
||||||
|
Only `storage_url` was checked, so after `s3.bucket` or `s3.prefix`
|
||||||
|
changed, the next backup treated every chunk as already stored,
|
||||||
|
uploaded nothing, and completed a snapshot that could not be restored.
|
||||||
|
Without `storage_url`, the local index is now bound to `s3.endpoint`,
|
||||||
|
`s3.bucket` and `s3.prefix`, and a change is refused as a changed
|
||||||
|
`storage_url` is.
|
||||||
|
|
||||||
|
- 2026-10-12: Rejected a retention duration longer than about 292 years
|
||||||
|
([issue #289](https://git.eeqj.de/sneak/vaultik/issues/289)). The
|
||||||
|
number given to `--older-than` or `--keep-newer-than` was multiplied
|
||||||
|
by the length of its unit with no limit, so from `293y`, `15251w`,
|
||||||
|
`3559mo` or `106752d`, or a sum of parts such as `200y200y`, the
|
||||||
|
result wrapped to a negative duration. The cutoff then lay in the
|
||||||
|
future, every snapshot of the name was deleted, the one just made
|
||||||
|
included, and the command exited 0. Such a duration is now rejected
|
||||||
|
as invalid and nothing is deleted.
|
||||||
|
|
||||||
|
- 2026-10-08: Made a backup cancelled under `--skip-errors` stop at once
|
||||||
|
([issue #286](https://git.eeqj.de/sneak/vaultik/issues/286)). After
|
||||||
|
Ctrl-C or SIGTERM, phase 2 treated the cancellation error from each
|
||||||
|
remaining file like an unreadable file: it opened the file, printed an
|
||||||
|
error line for it, counted it as failed and went on to the next one.
|
||||||
|
The run now stops at the first file after the cancellation, and no
|
||||||
|
file is counted as failed because of it.
|
||||||
|
|
||||||
|
- 2026-10-08: Made `remote nuke` delete the `.partial` files that
|
||||||
|
uploads cut off part-way leave on the destination store
|
||||||
|
([issue #281](https://git.eeqj.de/sneak/vaultik/issues/281)). The
|
||||||
|
file and rclone listings skip such a file, so the command left it in
|
||||||
|
place and still reported the store empty. It now removes them under
|
||||||
|
`metadata/` and `blobs/` as its last step.
|
||||||
|
|
||||||
|
- 2026-10-08: Made a local index error while recording a directory or
|
||||||
|
symlink stop a backup under `--skip-errors`
|
||||||
|
([issue #284](https://git.eeqj.de/sneak/vaultik/issues/284)). Phase 2
|
||||||
|
only records such an entry in the local index, since it has no data
|
||||||
|
to open or read, but an error doing so was skipped like an unreadable
|
||||||
|
file. The snapshot then completed without the entry, and a restore
|
||||||
|
did not recreate it. The error now stops the backup, as it does
|
||||||
|
without the flag.
|
||||||
|
|
||||||
- 2026-10-08: Counted a file that a backup could not store as failed
|
- 2026-10-08: Counted a file that a backup could not store as failed
|
||||||
([issue #280](https://git.eeqj.de/sneak/vaultik/issues/280)). A file
|
([issue #280](https://git.eeqj.de/sneak/vaultik/issues/280)). A file
|
||||||
that phase 1 counted and phase 2 could not open, because it was
|
that phase 1 counted and phase 2 could not open, because it was
|
||||||
|
|||||||
@@ -255,8 +255,12 @@ manifest (the destination store may hold snapshots this host doesn't
|
|||||||
know about), which is what 'vaultik prune' does. On success the command
|
know about), which is what 'vaultik prune' does. On success the command
|
||||||
prints the exact 'vaultik prune' invocation to run as a follow-up.
|
prints the exact 'vaultik prune' invocation to run as a follow-up.
|
||||||
|
|
||||||
|
An ID that names no snapshot in the local index or on the destination
|
||||||
|
store is an error.
|
||||||
|
|
||||||
Use --local-only to skip the remote half (e.g. when you want to forget a
|
Use --local-only to skip the remote half (e.g. when you want to forget a
|
||||||
snapshot locally without touching the destination store).
|
snapshot locally without touching the destination store); the snapshot
|
||||||
|
must then be in the local index.
|
||||||
|
|
||||||
If the remote is unreachable, the local-database removal still completes
|
If the remote is unreachable, the local-database removal still completes
|
||||||
and a warning is emitted; run 'vaultik snapshot remove <snapshot-id>' again
|
and a warning is emitted; run 'vaultik snapshot remove <snapshot-id>' again
|
||||||
|
|||||||
@@ -1312,6 +1312,13 @@ func (s *Scanner) processPhase(
|
|||||||
|
|
||||||
// Process each file
|
// Process each file
|
||||||
for _, fileToProcess := range filesToProcess {
|
for _, fileToProcess := range filesToProcess {
|
||||||
|
// Check context cancellation
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
// Update progress
|
// Update progress
|
||||||
if s.progress != nil {
|
if s.progress != nil {
|
||||||
s.progress.GetStats().CurrentFile.Store(fileToProcess.Path)
|
s.progress.GetStats().CurrentFile.Store(fileToProcess.Path)
|
||||||
@@ -1348,13 +1355,32 @@ func (s *Scanner) processPhase(
|
|||||||
return s.finalizeProcessPhase(ctx, result)
|
return s.finalizeProcessPhase(ctx, result)
|
||||||
}
|
}
|
||||||
|
|
||||||
// processFileWithErrorHandling wraps processFileStreaming with error recovery for
|
// processFileWithErrorHandling records a directory or symlink, or wraps
|
||||||
// deleted files and skip-errors mode. Returns (skipped, error).
|
// processFileStreaming for a regular file with error recovery for deleted
|
||||||
|
// files and skip-errors mode. Returns (skipped, error).
|
||||||
func (s *Scanner) processFileWithErrorHandling(
|
func (s *Scanner) processFileWithErrorHandling(
|
||||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||||
) (bool, error) {
|
) (bool, error) {
|
||||||
|
// A directory or symlink has no data to open or read; it is only
|
||||||
|
// recorded in the local index. An error recording it stops the run
|
||||||
|
// even under --skip-errors, or the snapshot would complete without it.
|
||||||
|
mode := os.FileMode(fileToProcess.File.Mode)
|
||||||
|
if mode&os.ModeSymlink != 0 || mode.IsDir() {
|
||||||
|
err := s.recordNonRegularFile(ctx, fileToProcess)
|
||||||
|
if err != nil {
|
||||||
|
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
err := s.processFileStreaming(ctx, fileToProcess, result)
|
err := s.processFileStreaming(ctx, fileToProcess, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// A cancelled run stops here even under --skip-errors, rather than
|
||||||
|
// counting this file as failed and going on to the next one.
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
|
||||||
|
}
|
||||||
// A packer/database/encryption/upload failure means the chunk's data
|
// A packer/database/encryption/upload failure means the chunk's data
|
||||||
// may not have been stored. Skipping the file would let the snapshot
|
// may not have been stored. Skipping the file would let the snapshot
|
||||||
// record a file whose chunk is in no blob and cannot be restored, so
|
// record a file whose chunk is in no blob and cannot be restored, so
|
||||||
@@ -1392,12 +1418,7 @@ func (s *Scanner) processFileWithErrorHandling(
|
|||||||
|
|
||||||
// countFailedFile counts a file that phase 2 could not store as failed
|
// countFailedFile counts a file that phase 2 could not store as failed
|
||||||
// and takes its size back out of BytesScanned, where phase 1 put it.
|
// and takes its size back out of BytesScanned, where phase 1 put it.
|
||||||
// Phase 1 counts no directories, so a directory is not counted here.
|
|
||||||
func countFailedFile(fileToProcess *FileToProcess, result *ScanResult) {
|
func countFailedFile(fileToProcess *FileToProcess, result *ScanResult) {
|
||||||
if fileToProcess.FileInfo.IsDir() {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
result.FilesFailed++
|
result.FilesFailed++
|
||||||
result.BytesScanned -= fileToProcess.FileInfo.Size()
|
result.BytesScanned -= fileToProcess.FileInfo.Size()
|
||||||
}
|
}
|
||||||
@@ -1770,16 +1791,11 @@ func (e *packerError) Error() string { return e.err.Error() }
|
|||||||
|
|
||||||
func (e *packerError) Unwrap() error { return e.err }
|
func (e *packerError) Unwrap() error { return e.err }
|
||||||
|
|
||||||
// processFileStreaming processes a file by streaming chunks directly to the packer
|
// processFileStreaming processes a regular file by streaming chunks directly
|
||||||
|
// to the packer
|
||||||
func (s *Scanner) processFileStreaming(
|
func (s *Scanner) processFileStreaming(
|
||||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||||
) error {
|
) error {
|
||||||
// Symlinks and directories have no data to chunk — just record them in the DB.
|
|
||||||
mode := os.FileMode(fileToProcess.File.Mode)
|
|
||||||
if mode&os.ModeSymlink != 0 || mode.IsDir() {
|
|
||||||
return s.recordNonRegularFile(ctx, fileToProcess)
|
|
||||||
}
|
|
||||||
|
|
||||||
file, err := s.fs.Open(fileToProcess.Path)
|
file, err := s.fs.Open(fileToProcess.Path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("opening file: %w", wrapPermissionError(fileToProcess.Path, err))
|
return fmt.Errorf("opening file: %w", wrapPermissionError(fileToProcess.Path, err))
|
||||||
|
|||||||
@@ -82,6 +82,32 @@ func (f *readFailFs) Open(name string) (afero.File, error) {
|
|||||||
return file, nil
|
return file, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// cancelOnOpenFs cancels the run when the scanner opens the target file to
|
||||||
|
// back it up, as Ctrl-C partway through processing would, and records each
|
||||||
|
// file opened after that.
|
||||||
|
type cancelOnOpenFs struct {
|
||||||
|
afero.Fs
|
||||||
|
|
||||||
|
target string
|
||||||
|
cancel context.CancelFunc
|
||||||
|
cancelled bool
|
||||||
|
openedAfterCancel []string
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:ireturn // afero.Fs.Open is defined to return the interface.
|
||||||
|
func (f *cancelOnOpenFs) Open(name string) (afero.File, error) {
|
||||||
|
if f.cancelled {
|
||||||
|
f.openedAfterCancel = append(f.openedAfterCancel, name)
|
||||||
|
}
|
||||||
|
|
||||||
|
if name == f.target {
|
||||||
|
f.cancel()
|
||||||
|
f.cancelled = true
|
||||||
|
}
|
||||||
|
|
||||||
|
return f.Fs.Open(name)
|
||||||
|
}
|
||||||
|
|
||||||
// linkRemovedAfterLstatFs is the real filesystem, except that the symlink at
|
// linkRemovedAfterLstatFs is the real filesystem, except that the symlink at
|
||||||
// target is removed right after the walk lstats it, as happens when a link is
|
// target is removed right after the walk lstats it, as happens when a link is
|
||||||
// deleted during a backup. The scanner's readlink of it then fails.
|
// deleted during a backup. The scanner's readlink of it then fails.
|
||||||
@@ -128,15 +154,16 @@ func writeSkipErrorTestFile(t *testing.T, fs afero.Fs, path, content string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// runSkipErrorScan scans source on fs with the given skip-errors setting,
|
// runSkipErrorScan scans source on fs under ctx with the given skip-errors
|
||||||
// printing user-facing messages to uiw (nil discards them), and returns the
|
// setting, printing user-facing messages to uiw (nil discards them), and
|
||||||
// repositories (for inspection) and the scan error.
|
// returns the repositories (for inspection) and the scan error.
|
||||||
func runSkipErrorScan(
|
func runSkipErrorScan(
|
||||||
t *testing.T, fs afero.Fs, source string, skipErrors bool, uiw *ui.Writer,
|
ctx context.Context, t *testing.T, fs afero.Fs, source string,
|
||||||
|
skipErrors bool, uiw *ui.Writer,
|
||||||
) (*database.Repositories, error) {
|
) (*database.Repositories, error) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
db, err := database.NewTestDB()
|
db, err := database.New(ctx, ":memory:")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("create test db: %v", err)
|
t.Fatalf("create test db: %v", err)
|
||||||
}
|
}
|
||||||
@@ -161,7 +188,6 @@ func runSkipErrorScan(
|
|||||||
SkipErrors: skipErrors,
|
SkipErrors: skipErrors,
|
||||||
})
|
})
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
snapshotID := "test-snapshot-skip-errors"
|
snapshotID := "test-snapshot-skip-errors"
|
||||||
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
||||||
|
|
||||||
@@ -185,7 +211,7 @@ func TestScannerPackingFailureAbortsUnderSkipErrors(t *testing.T) {
|
|||||||
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
||||||
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatal("expected scan to abort on the packer error, got nil")
|
t.Fatal("expected scan to abort on the packer error, got nil")
|
||||||
}
|
}
|
||||||
@@ -212,7 +238,7 @@ func TestScannerReadErrorAbortsWithoutSkipErrors(t *testing.T) {
|
|||||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||||
|
|
||||||
_, err := runSkipErrorScan(t, fs, "/source", false, nil)
|
_, err := runSkipErrorScan(context.Background(), t, fs, "/source", false, nil)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatal("expected scan to fail on the read error, got nil")
|
t.Fatal("expected scan to fail on the read error, got nil")
|
||||||
}
|
}
|
||||||
@@ -228,7 +254,7 @@ func TestScannerReadErrorSkippedWithSkipErrors(t *testing.T) {
|
|||||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -267,7 +293,7 @@ func TestScannerUnreadableSymlinkAbortsWithoutSkipErrors(t *testing.T) {
|
|||||||
sourceDir, linkPath := writeSymlinkSource(t)
|
sourceDir, linkPath := writeSymlinkSource(t)
|
||||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||||
|
|
||||||
_, err := runSkipErrorScan(t, fs, sourceDir, false, nil)
|
_, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, false, nil)
|
||||||
if !errors.Is(err, os.ErrNotExist) {
|
if !errors.Is(err, os.ErrNotExist) {
|
||||||
t.Fatalf("expected scan to fail on the removed symlink, got %v", err)
|
t.Fatalf("expected scan to fail on the removed symlink, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -283,7 +309,7 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
|||||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||||
uiw := ui.NewWithColor(io.Discard, false)
|
uiw := ui.NewWithColor(io.Discard, false)
|
||||||
|
|
||||||
repos, err := runSkipErrorScan(t, fs, sourceDir, true, uiw)
|
repos, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, true, uiw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||||
}
|
}
|
||||||
@@ -302,3 +328,56 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
|||||||
t.Fatalf("expected %s not to be recorded", linkPath)
|
t.Fatalf("expected %s not to be recorded", linkPath)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestScannerCancelStopsSkipErrorsRun checks that a --skip-errors backup
|
||||||
|
// cancelled partway through processing returns the cancellation error,
|
||||||
|
// opens no further file, and reports no file as failed.
|
||||||
|
func TestScannerCancelStopsSkipErrorsRun(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
targetContent string
|
||||||
|
}{
|
||||||
|
// The cancellation lands while the target is being read.
|
||||||
|
{name: "while reading a file", targetContent: "first file content"},
|
||||||
|
// An empty target has no chunks, so the cancellation goes unnoticed
|
||||||
|
// until the run moves on to the next file.
|
||||||
|
{name: "between files", targetContent: ""},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
const target = "/source/a.txt"
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
fs := &cancelOnOpenFs{
|
||||||
|
Fs: afero.NewMemMapFs(), target: target, cancel: cancel,
|
||||||
|
}
|
||||||
|
writeSkipErrorTestFile(t, fs, target, tt.targetContent)
|
||||||
|
writeSkipErrorTestFile(t, fs, "/source/b.txt", "second file content")
|
||||||
|
writeSkipErrorTestFile(t, fs, "/source/c.txt", "third file content")
|
||||||
|
|
||||||
|
uiw := ui.NewWithColor(io.Discard, false)
|
||||||
|
|
||||||
|
_, err := runSkipErrorScan(ctx, t, fs, "/source", true, uiw)
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("expected the cancellation error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(fs.openedAfterCancel) != 0 {
|
||||||
|
t.Fatalf("expected no file opened after the cancellation, got %v",
|
||||||
|
fs.openedAfterCancel)
|
||||||
|
}
|
||||||
|
|
||||||
|
if uiw.ErrorCount() != 0 {
|
||||||
|
t.Fatalf("expected no file reported as failed, got %d error lines",
|
||||||
|
uiw.ErrorCount())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,15 +1,16 @@
|
|||||||
// Package faultstore provides a storage.Storer wrapper that injects
|
// Package faultstore provides a storage.Storer wrapper that injects
|
||||||
// faults on demand, so tests can reproduce the failure modes a real
|
// faults on demand, so tests can reproduce the failure modes a real
|
||||||
// backend exhibits: an upload that fails partway, a backend that reports
|
// backend exhibits: an upload that fails partway, a backend that reports
|
||||||
// success while storing nothing, and reads that return corrupt or
|
// success while storing nothing, reads that return corrupt or truncated
|
||||||
// truncated bytes. It is the seam called for by the fault-injection
|
// bytes, and a delete the backend refuses. It is the seam called for by
|
||||||
// tests (sneak/vaultik issue 72) and is meant to be reused by future
|
// the fault-injection tests (sneak/vaultik issue 72) and is meant to be
|
||||||
// tests rather than re-implemented per case.
|
// reused by future tests rather than re-implemented per case.
|
||||||
//
|
//
|
||||||
// The wrapper delegates every method to the inner Storer. Two hooks
|
// The wrapper delegates every method to the inner Storer. Three hooks
|
||||||
// change that: OnPut decides the fate of each write, and OnGet decides
|
// change that: OnPut decides the fate of each write, OnGet decides how
|
||||||
// how each read's bytes are returned. Both are keyed by the object key,
|
// each read's bytes are returned, and OnDelete decides whether each
|
||||||
// so a test can fault only blobs, only metadata, or a single object.
|
// delete fails. All are keyed by the object key, so a test can fault
|
||||||
|
// only blobs, only metadata, or a single object.
|
||||||
package faultstore
|
package faultstore
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -25,6 +26,10 @@ import (
|
|||||||
// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
|
// ErrInjectedUpload is returned by a Put the OnPut hook chose to fail.
|
||||||
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")
|
var ErrInjectedUpload = errors.New("faultstore: injected upload failure")
|
||||||
|
|
||||||
|
// ErrInjectedDelete is returned by a Delete the OnDelete hook chose to
|
||||||
|
// fail.
|
||||||
|
var ErrInjectedDelete = errors.New("faultstore: injected delete failure")
|
||||||
|
|
||||||
// PutAction is the disposition OnPut assigns to a write.
|
// PutAction is the disposition OnPut assigns to a write.
|
||||||
type PutAction int
|
type PutAction int
|
||||||
|
|
||||||
@@ -66,6 +71,11 @@ type Storer struct {
|
|||||||
// OnGet, when set, is consulted for every Get with the object key
|
// OnGet, when set, is consulted for every Get with the object key
|
||||||
// and damages the returned bytes accordingly.
|
// and damages the returned bytes accordingly.
|
||||||
OnGet func(key string) GetFault
|
OnGet func(key string) GetFault
|
||||||
|
|
||||||
|
// OnDelete, when set, is consulted before every Delete with the
|
||||||
|
// object key. Returning true fails the delete and leaves the object
|
||||||
|
// in place.
|
||||||
|
OnDelete func(key string) bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// New wraps inner. inner must be non-nil.
|
// New wraps inner. inner must be non-nil.
|
||||||
@@ -154,8 +164,12 @@ func (f *Storer) Stat(ctx context.Context, key string) (*storage.ObjectInfo, err
|
|||||||
return f.inner.Stat(ctx, key)
|
return f.inner.Stat(ctx, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Delete delegates unchanged.
|
// Delete removes the object unless OnDelete fails the delete.
|
||||||
func (f *Storer) Delete(ctx context.Context, key string) error {
|
func (f *Storer) Delete(ctx context.Context, key string) error {
|
||||||
|
if f.OnDelete != nil && f.OnDelete(key) {
|
||||||
|
return fmt.Errorf("%w for %q", ErrInjectedDelete, key)
|
||||||
|
}
|
||||||
|
|
||||||
return f.inner.Delete(ctx, key)
|
return f.inner.Delete(ctx, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -171,6 +185,11 @@ func (f *Storer) ListStream(
|
|||||||
return f.inner.ListStream(ctx, prefix)
|
return f.inner.ListStream(ctx, prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads delegates unchanged.
|
||||||
|
func (f *Storer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
return f.inner.DeletePartialUploads(ctx, prefix)
|
||||||
|
}
|
||||||
|
|
||||||
// Info delegates unchanged.
|
// Info delegates unchanged.
|
||||||
func (f *Storer) Info() storage.Info {
|
func (f *Storer) Info() storage.Info {
|
||||||
return f.inner.Info()
|
return f.inner.Info()
|
||||||
|
|||||||
@@ -242,6 +242,42 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every file under prefix whose name ends in
|
||||||
|
// tempSuffix. A missing prefix has none to remove.
|
||||||
|
func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
|
exists, err := afero.Exists(f.fs, basePath)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("checking path: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !exists {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
if info.IsDir() || !strings.HasSuffix(info.Name(), tempSuffix) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return f.fs.Remove(path)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("walking directory: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (f *FileStorer) Info() Info {
|
func (f *FileStorer) Info() Info {
|
||||||
return Info{
|
return Info{
|
||||||
|
|||||||
@@ -120,3 +120,45 @@ func TestFileStorer_ListSkipsPartialFiles(t *testing.T) {
|
|||||||
t.Fatalf("ListStream should return only the real key, got %v", streamed)
|
t.Fatalf("ListStream should return only the real key, got %v", streamed)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestFileStorer_DeletePartialUploads checks that a leftover temp file is
|
||||||
|
// removed and the object at the real key is kept.
|
||||||
|
func TestFileStorer_DeletePartialUploads(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
base := t.TempDir()
|
||||||
|
|
||||||
|
f, err := storage.NewFileStorer(base)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewFileStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
err = f.Put(ctx, testBlobKey, strings.NewReader("blob-bytes"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
leftover := filepath.Join(base, testBlobKey+"-123456.partial")
|
||||||
|
|
||||||
|
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("writing leftover temp file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = f.DeletePartialUploads(ctx, "blobs/")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DeletePartialUploads: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = os.Stat(leftover)
|
||||||
|
if !os.IsNotExist(err) {
|
||||||
|
t.Errorf("leftover temp file was not removed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = f.Stat(ctx, testBlobKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("Stat of the real key: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -213,6 +213,31 @@ func (r *RcloneStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every object under prefix whose name ends
|
||||||
|
// in tempSuffix.
|
||||||
|
func (r *RcloneStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
|
||||||
|
var partial []fs.Object
|
||||||
|
|
||||||
|
err := operations.ListFn(ctx, r.fsys, func(obj fs.Object) {
|
||||||
|
key := obj.Remote()
|
||||||
|
if strings.HasPrefix(key, prefix) && strings.HasSuffix(key, tempSuffix) {
|
||||||
|
partial = append(partial, obj)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("listing objects: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, obj := range partial {
|
||||||
|
err = obj.Remove(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("removing object: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (r *RcloneStorer) Info() Info {
|
func (r *RcloneStorer) Info() Info {
|
||||||
location := r.remote
|
location := r.remote
|
||||||
|
|||||||
@@ -198,6 +198,47 @@ func TestRcloneStorerListSkipsPartialFiles(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRcloneStorerDeletePartialUploads checks that a temporary file left
|
||||||
|
// by a killed upload is removed and the object at the real key is kept.
|
||||||
|
//
|
||||||
|
//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
|
||||||
|
func TestRcloneStorerDeletePartialUploads(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
s, err := storage.NewRcloneStorer(ctx, ":local", dir)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewRcloneStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.Put(ctx, testBlobKey, strings.NewReader("blob-bytes"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
leftover := filepath.Join(dir, testBlobKey+"-123456.partial")
|
||||||
|
|
||||||
|
err = os.WriteFile(leftover, []byte("half"), 0o600)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("writing leftover temp file: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.DeletePartialUploads(ctx, "blobs/")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DeletePartialUploads: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = os.Stat(leftover)
|
||||||
|
if !os.IsNotExist(err) {
|
||||||
|
t.Errorf("leftover temp file was not removed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = s.Stat(ctx, testBlobKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("Stat of the real key: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// newRcloneStorerOnWrappedLocal registers name as rclone's local backend
|
// newRcloneStorerOnWrappedLocal registers name as rclone's local backend
|
||||||
// wrapped by wrap, and builds an rclone backend on it rooted at a fresh
|
// wrapped by wrap, and builds an rclone backend on it rooted at a fresh
|
||||||
// temp directory. wrap changes the features the local backend reports, so
|
// temp directory. wrap changes the features the local backend reports, so
|
||||||
|
|||||||
@@ -99,6 +99,12 @@ func (s *S3Storer) ListStream(ctx context.Context, prefix string) <-chan ObjectI
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: S3 shows an object only once
|
||||||
|
// its upload has completed, so an upload cut off part-way leaves none.
|
||||||
|
func (s *S3Storer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
func (s *S3Storer) Info() Info {
|
func (s *S3Storer) Info() Info {
|
||||||
return Info{
|
return Info{
|
||||||
|
|||||||
@@ -71,6 +71,12 @@ type Storer interface {
|
|||||||
// If an error occurs during listing, the final item will have Err set.
|
// If an error occurs during listing, the final item will have Err set.
|
||||||
ListStream(ctx context.Context, prefix string) <-chan ObjectInfo
|
ListStream(ctx context.Context, prefix string) <-chan ObjectInfo
|
||||||
|
|
||||||
|
// DeletePartialUploads removes every object under prefix that an
|
||||||
|
// upload cut off part-way left under a temporary name ending in
|
||||||
|
// `.partial`. The file and rclone backends' List and ListStream skip
|
||||||
|
// such an object.
|
||||||
|
DeletePartialUploads(ctx context.Context, prefix string) error
|
||||||
|
|
||||||
// Info returns human-readable storage location information.
|
// Info returns human-readable storage location information.
|
||||||
Info() Info
|
Info() Info
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -257,6 +257,10 @@ func (s *stubLister) List(_ context.Context, _ string) ([]string, error) {
|
|||||||
return nil, errStubUnused
|
return nil, errStubUnused
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *stubLister) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return errStubUnused
|
||||||
|
}
|
||||||
|
|
||||||
func (s *stubLister) Info() storage.Info {
|
func (s *stubLister) Info() storage.Info {
|
||||||
return storage.Info{}
|
return storage.Info{}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package vaultik
|
|||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -47,6 +48,10 @@ const (
|
|||||||
year = 365 * day
|
year = 365 * day
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// maxDuration is the largest value a time.Duration can hold, about 292
|
||||||
|
// years.
|
||||||
|
const maxDuration = time.Duration(math.MaxInt64)
|
||||||
|
|
||||||
// A snapshot ID split on "_" has at least a hostname and a trailing
|
// A snapshot ID split on "_" has at least a hostname and a trailing
|
||||||
// timestamp.
|
// timestamp.
|
||||||
const minSnapshotIDParts = 2
|
const minSnapshotIDParts = 2
|
||||||
@@ -145,10 +150,10 @@ func parseSnapshotName(snapshotID, hostname string) string {
|
|||||||
// parseDuration parses a duration string with support for human-friendly units:
|
// parseDuration parses a duration string with support for human-friendly units:
|
||||||
// d/day/days, w/week/weeks, mo/month/months, y/year/years, plus standard Go
|
// d/day/days, w/week/weeks, mo/month/months, y/year/years, plus standard Go
|
||||||
// duration units. Following Go, m is minutes and mo is months. A bare number,
|
// duration units. Following Go, m is minutes and mo is months. A bare number,
|
||||||
// an unknown unit, and a negative value are all rejected. Outside the Go
|
// an unknown unit, a negative value, and a total longer than maxDuration are
|
||||||
// units the input must be whole numbers each followed directly by a unit
|
// all rejected. Outside the Go units the input must be whole numbers each
|
||||||
// (2w3d), so 1.0y and 30 days are errors; decimals work only in Go units
|
// followed directly by a unit (2w3d), so 1.0y and 30 days are errors;
|
||||||
// (1.5h).
|
// decimals work only in Go units (1.5h).
|
||||||
func parseDuration(s string) (time.Duration, error) {
|
func parseDuration(s string) (time.Duration, error) {
|
||||||
if strings.HasPrefix(strings.TrimSpace(s), "-") {
|
if strings.HasPrefix(strings.TrimSpace(s), "-") {
|
||||||
return 0, errNegativeDuration
|
return 0, errNegativeDuration
|
||||||
@@ -181,19 +186,30 @@ func parseDuration(s string) (time.Duration, error) {
|
|||||||
return 0, fmt.Errorf("invalid number %q: %w", match[1], err)
|
return 0, fmt.Errorf("invalid number %q: %w", match[1], err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var unitLength time.Duration
|
||||||
|
|
||||||
unit := strings.ToLower(match[2])
|
unit := strings.ToLower(match[2])
|
||||||
switch unit {
|
switch unit {
|
||||||
case "d", "day", "days":
|
case "d", "day", "days":
|
||||||
total += time.Duration(n) * day
|
unitLength = day
|
||||||
case "w", "week", "weeks":
|
case "w", "week", "weeks":
|
||||||
total += time.Duration(n) * week
|
unitLength = week
|
||||||
case "mo", "month", "months":
|
case "mo", "month", "months":
|
||||||
total += time.Duration(n) * month
|
unitLength = month
|
||||||
case "y", "year", "years":
|
case "y", "year", "years":
|
||||||
total += time.Duration(n) * year
|
unitLength = year
|
||||||
default:
|
default:
|
||||||
return 0, fmt.Errorf("%w %q", errUnknownTimeUnit, unit)
|
return 0, fmt.Errorf("%w %q", errUnknownTimeUnit, unit)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Past maxDuration the sum wraps to a negative value, which would
|
||||||
|
// put the purge cutoff in the future.
|
||||||
|
if time.Duration(n) > (maxDuration-total)/unitLength {
|
||||||
|
return 0, fmt.Errorf("%w: %q is longer than about 292 years",
|
||||||
|
errInvalidDuration, s)
|
||||||
|
}
|
||||||
|
|
||||||
|
total += time.Duration(n) * unitLength
|
||||||
}
|
}
|
||||||
|
|
||||||
return total, nil
|
return total, nil
|
||||||
|
|||||||
@@ -111,6 +111,17 @@ func TestParseDuration(t *testing.T) {
|
|||||||
{"30 days", 0, true},
|
{"30 days", 0, true},
|
||||||
{"30 days ago", 0, true},
|
{"30 days ago", 0, true},
|
||||||
{"x7d", 0, true},
|
{"x7d", 0, true},
|
||||||
|
// The largest time.Duration is about 292 years. A larger value
|
||||||
|
// must be rejected, not wrapped to a negative one, which would
|
||||||
|
// put the cutoff in the future and select every snapshot.
|
||||||
|
{"292y", 292 * 365 * 24 * time.Hour, false},
|
||||||
|
{"293y", 0, true},
|
||||||
|
{"3559mo", 0, true},
|
||||||
|
{"15251w", 0, true},
|
||||||
|
{"106752d", 0, true},
|
||||||
|
{"2562048h", 0, true},
|
||||||
|
{"200y200y", 0, true},
|
||||||
|
{"292y6mo", 0, true},
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
|
|||||||
@@ -155,6 +155,12 @@ func (m *MockStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: Put stores each object
|
||||||
|
// under its key at once.
|
||||||
|
func (m *MockStorer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (m *MockStorer) Info() storage.Info {
|
func (m *MockStorer) Info() storage.Info {
|
||||||
return storage.Info{
|
return storage.Info{
|
||||||
Type: "mock",
|
Type: "mock",
|
||||||
|
|||||||
@@ -0,0 +1,144 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io/fs"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestNukeRemoteLeavesNoFiles checks that remote nuke leaves no file
|
||||||
|
// under a file:// destination, including the `.partial` files uploads
|
||||||
|
// killed part-way leave next to a snapshot's metadata and next to where a
|
||||||
|
// blob would have been.
|
||||||
|
func TestNukeRemoteLeavesNoFiles(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, repos, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, snapshots, 1)
|
||||||
|
|
||||||
|
snapshotKey := snapshot.RemoteSnapshotKey(snapshots[0].ID.String())
|
||||||
|
hash := testBlobHashA
|
||||||
|
leftovers := []string{
|
||||||
|
filepath.Join(storeDir, "metadata", snapshotKey,
|
||||||
|
"db.zst.age-123456.partial"),
|
||||||
|
filepath.Join(storeDir, "blobs", hash[:2], hash[2:4],
|
||||||
|
hash+"-123456.partial"),
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, leftover := range leftovers {
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Dir(leftover), 0o750))
|
||||||
|
require.NoError(t, os.WriteFile(leftover, []byte("half an upload"), 0o600))
|
||||||
|
}
|
||||||
|
|
||||||
|
require.NoError(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
var files []string
|
||||||
|
|
||||||
|
err = filepath.WalkDir(storeDir,
|
||||||
|
func(path string, entry fs.DirEntry, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if !entry.IsDir() {
|
||||||
|
files = append(files, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Empty(t, files)
|
||||||
|
|
||||||
|
// Nuke deletes the blobs itself, so advice to run prune is wrong.
|
||||||
|
assert.Contains(t, out.String(), "Backup destination store is now empty.")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik prune")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteFailsWhenADeleteFails checks that remote nuke returns an
|
||||||
|
// error when a snapshot's metadata or a blob cannot be deleted, does not
|
||||||
|
// report the store empty, and advises running nuke again: prune deletes
|
||||||
|
// no metadata.
|
||||||
|
func TestNukeRemoteFailsWhenADeleteFails(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, refusedPrefix := range []string{"metadata/", "blobs/"} {
|
||||||
|
t.Run(refusedPrefix, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
store := faultstore.New(v.Storage)
|
||||||
|
store.OnDelete = func(key string) bool {
|
||||||
|
return strings.HasPrefix(key, refusedPrefix)
|
||||||
|
}
|
||||||
|
v.Storage = store
|
||||||
|
|
||||||
|
require.Error(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
assert.NotContains(t, out.String(), "now empty")
|
||||||
|
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik prune")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice checks that
|
||||||
|
// remote nuke, when the local index belongs to a different destination
|
||||||
|
// than the configured one, fails without advising to run nuke again:
|
||||||
|
// that run would fail the same way.
|
||||||
|
func TestNukeRemoteRefusesAnotherDestinationWithoutRetryAdvice(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
v.Config.StorageURL = "file://" + filepath.Join(t.TempDir(), "elsewhere")
|
||||||
|
|
||||||
|
err := v.NukeRemote(true)
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "bound to a different backup destination")
|
||||||
|
assert.NotContains(t, out.String(), "vaultik remote nuke --force")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved checks that a
|
||||||
|
// snapshot remote nuke cannot remove from the local index fails the
|
||||||
|
// run: nuke then leaves that snapshot's metadata on the destination
|
||||||
|
// store.
|
||||||
|
func TestNukeRemoteFailsWhenALocalSnapshotCannotBeRemoved(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, out := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
_, err := v.DB.Conn().ExecContext(ctx, `
|
||||||
|
CREATE TRIGGER refuse_snapshot_delete BEFORE DELETE ON snapshots
|
||||||
|
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
require.Error(t, v.NukeRemote(true))
|
||||||
|
|
||||||
|
assert.NotContains(t, out.String(), "now empty")
|
||||||
|
assert.Contains(t, out.String(), "vaultik remote nuke --force")
|
||||||
|
}
|
||||||
@@ -19,13 +19,22 @@ type PruneOptions struct {
|
|||||||
var errNukeRequiresForce = errors.New(
|
var errNukeRequiresForce = errors.New(
|
||||||
"nuke requires --force (this deletes ALL remote snapshots and blobs)")
|
"nuke requires --force (this deletes ALL remote snapshots and blobs)")
|
||||||
|
|
||||||
|
// errBlobDeleteFailed reports unreferenced blobs prune could not delete.
|
||||||
|
var errBlobDeleteFailed = errors.New("blob(s) could not be deleted")
|
||||||
|
|
||||||
|
// nukeCommandHint is the command suggested when a remote nuke could not
|
||||||
|
// delete everything: running it again deletes what is left, which
|
||||||
|
// `vaultik prune` does not do for snapshot metadata.
|
||||||
|
const nukeCommandHint = "vaultik remote nuke --force"
|
||||||
|
|
||||||
// metadataDirName is the top-level remote directory holding snapshot
|
// metadataDirName is the top-level remote directory holding snapshot
|
||||||
// metadata.
|
// metadata.
|
||||||
const metadataDirName = "metadata"
|
const metadataDirName = "metadata"
|
||||||
|
|
||||||
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
||||||
// storage. After this returns successfully the bucket prefix is empty and
|
// storage, along with any object an upload cut off part-way left under a
|
||||||
// the next backup starts from scratch.
|
// temporary `.partial` name. After this returns successfully the bucket
|
||||||
|
// prefix is empty and the next backup starts from scratch.
|
||||||
//
|
//
|
||||||
// Refuses to run unless force is true. The caller is responsible for
|
// Refuses to run unless force is true. The caller is responsible for
|
||||||
// confirming with the user.
|
// confirming with the user.
|
||||||
@@ -34,6 +43,29 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
|||||||
return errNukeRequiresForce
|
return errNukeRequiresForce
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Checked before anything is deleted: running nuke again would fail
|
||||||
|
// the same way, so the advice below must not follow this error.
|
||||||
|
err := v.EnsureStorageBinding()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = v.emptyDestinationStore()
|
||||||
|
if err != nil {
|
||||||
|
v.UI.Warningf("Run '%s' again to delete what is left on the "+
|
||||||
|
"backup destination store.", nukeCommandHint)
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
v.UI.Completef("Backup destination store is now empty.")
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// emptyDestinationStore does the deleting for NukeRemote and stops at
|
||||||
|
// the first step that fails.
|
||||||
|
func (v *Vaultik) emptyDestinationStore() error {
|
||||||
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
v.UI.Beginf("Removing all snapshot metadata from backup destination store.")
|
||||||
|
|
||||||
_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
|
_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
|
||||||
@@ -48,7 +80,14 @@ func (v *Vaultik) NukeRemote(force bool) error {
|
|||||||
return fmt.Errorf("pruning blobs: %w", err)
|
return fmt.Errorf("pruning blobs: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
v.UI.Completef("Backup destination store is now empty.")
|
// The file and rclone listings skip `.partial` objects, so the two
|
||||||
|
// steps above never delete them.
|
||||||
|
for _, prefix := range []string{"metadata/", "blobs/"} {
|
||||||
|
err = v.Storage.DeletePartialUploads(v.ctx, prefix)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("deleting partial uploads: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -152,14 +191,17 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
|
|||||||
v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
|
v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return v.outputPruneBlobsJSON(result)
|
err = v.outputPruneBlobsJSON(result)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
v.stdoutf("\nDeleted %d blob(s) totaling %s\n",
|
||||||
result.BlobsDeleted, ubytes(result.BytesFreed))
|
result.BlobsDeleted, ubytes(result.BytesFreed))
|
||||||
|
}
|
||||||
|
|
||||||
if result.BlobsFailed > 0 {
|
if result.BlobsFailed > 0 {
|
||||||
v.stdoutf("Failed to delete %d blob(s)\n", result.BlobsFailed)
|
return fmt.Errorf("%d %w", result.BlobsFailed, errBlobDeleteFailed)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -0,0 +1,98 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage/faultstore"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// refuseBlobDeletes is an OnDelete hook that fails every blob delete.
|
||||||
|
func refuseBlobDeletes(key string) bool {
|
||||||
|
return strings.HasPrefix(key, "blobs/")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPruneBlobs_FailedDeleteReturnsError checks that prune returns an
|
||||||
|
// error, so the command exits non-zero, when an unreferenced blob cannot
|
||||||
|
// be deleted.
|
||||||
|
func TestPruneBlobs_FailedDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
|
||||||
|
// The store holds no manifest, so nothing references this blob.
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
|
||||||
|
store := faultstore.New(env.store)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
env.v.Storage = store
|
||||||
|
|
||||||
|
err := env.v.PruneBlobs(&vaultik.PruneOptions{Force: true})
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPruneBlobs_JSONFailedDeleteReturnsError checks that prune --json
|
||||||
|
// prints its document and then returns an error when an unreferenced
|
||||||
|
// blob cannot be deleted.
|
||||||
|
func TestPruneBlobs_JSONFailedDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
|
||||||
|
store := faultstore.New(env.store)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
env.v.Storage = store
|
||||||
|
|
||||||
|
err := env.v.PruneBlobs(&vaultik.PruneOptions{JSON: true})
|
||||||
|
|
||||||
|
require.Error(t, err, "prune must fail when a blob cannot be deleted")
|
||||||
|
|
||||||
|
var result vaultik.PruneBlobsResult
|
||||||
|
|
||||||
|
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &result),
|
||||||
|
"stdout must hold the JSON document, got:\n%s", env.stdout.String())
|
||||||
|
assert.Equal(t, 1, result.BlobsFailed)
|
||||||
|
assert.Equal(t, 0, result.BlobsDeleted)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestCreateSnapshotPrune_FailedBlobDeleteReturnsError checks that the
|
||||||
|
// prune step of snapshot create --prune fails the run when an
|
||||||
|
// unreferenced blob cannot be deleted.
|
||||||
|
func TestCreateSnapshotPrune_FailedBlobDeleteReturnsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, _, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
|
||||||
|
// No snapshot references this blob, so the prune step deletes it.
|
||||||
|
orphan := filepath.Join(storeDir, "blobs", testBlobHashA[:2],
|
||||||
|
testBlobHashA[2:4], testBlobHashA)
|
||||||
|
require.NoError(t, os.MkdirAll(filepath.Dir(orphan), 0o750))
|
||||||
|
require.NoError(t, os.WriteFile(orphan, []byte("orphan"), 0o600))
|
||||||
|
|
||||||
|
store := faultstore.New(v.Storage)
|
||||||
|
store.OnDelete = refuseBlobDeletes
|
||||||
|
v.Storage = store
|
||||||
|
|
||||||
|
err := v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
Cron: true,
|
||||||
|
Prune: true,
|
||||||
|
})
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "1 blob(s) could not be deleted")
|
||||||
|
assert.FileExists(t, orphan)
|
||||||
|
}
|
||||||
@@ -267,6 +267,29 @@ func TestPurgeOlderThan_WithNameFilter(t *testing.T) {
|
|||||||
assert.Contains(t, remaining, snapHomeT0)
|
assert.Contains(t, remaining, snapHomeT0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPurgeOlderThan_TooLargeDeletesNothing(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
snapshotIDs := []string{snapHomeT0, snapHomeT1}
|
||||||
|
|
||||||
|
v := setupPurgeTest(t, snapHostname, snapshotIDs)
|
||||||
|
|
||||||
|
err := v.PurgeSnapshotsWithOptions(&vaultik.SnapshotPurgeOptions{
|
||||||
|
OlderThan: "1000y",
|
||||||
|
Force: true,
|
||||||
|
})
|
||||||
|
require.Error(t, err)
|
||||||
|
|
||||||
|
err = v.PurgeSnapshotsWithOptions(&vaultik.SnapshotPurgeOptions{
|
||||||
|
OlderThan: "292y",
|
||||||
|
Force: true,
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
assert.ElementsMatch(t, snapshotIDs, listRemainingSnapshots(t, v))
|
||||||
|
}
|
||||||
|
|
||||||
func TestPurgeKeepLatest_ThreeNames(t *testing.T) {
|
func TestPurgeKeepLatest_ThreeNames(t *testing.T) {
|
||||||
log.Initialize(log.Config{})
|
log.Initialize(log.Config{})
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|||||||
@@ -3,17 +3,21 @@ package vaultik_test
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
|
"database/sql"
|
||||||
"io"
|
"io"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/klauspost/compress/zstd"
|
"github.com/klauspost/compress/zstd"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
"sneak.berlin/go/vaultik/internal/log"
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/types"
|
||||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -125,6 +129,12 @@ func (s *testStorer) ListStream(
|
|||||||
return ch
|
return ch
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeletePartialUploads has nothing to remove: Put stores each object
|
||||||
|
// under its key at once.
|
||||||
|
func (s *testStorer) DeletePartialUploads(_ context.Context, _ string) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (s *testStorer) Info() storage.Info {
|
func (s *testStorer) Info() storage.Info {
|
||||||
return storage.Info{
|
return storage.Info{
|
||||||
Type: testLabel,
|
Type: testLabel,
|
||||||
@@ -208,6 +218,37 @@ func addBlob(t *testing.T, store *testStorer, hash string) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// newLocalIndex returns an in-memory local index holding a completed
|
||||||
|
// snapshot row for each of ids.
|
||||||
|
func newLocalIndex(t *testing.T, ids ...string) *database.Repositories {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
db, err := database.New(ctx, ":memory:")
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
|
||||||
|
for _, id := range ids {
|
||||||
|
completedAt := time.Now().UTC()
|
||||||
|
snap := &database.Snapshot{
|
||||||
|
ID: types.SnapshotID(id),
|
||||||
|
Hostname: snapHostname,
|
||||||
|
StartedAt: completedAt,
|
||||||
|
CompletedAt: &completedAt,
|
||||||
|
}
|
||||||
|
|
||||||
|
err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
|
return repos.Snapshots.Create(ctx, tx, snap)
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return repos
|
||||||
|
}
|
||||||
|
|
||||||
// ============================================================================
|
// ============================================================================
|
||||||
// Unit Tests for RemoveSnapshot
|
// Unit Tests for RemoveSnapshot
|
||||||
// ============================================================================
|
// ============================================================================
|
||||||
@@ -227,6 +268,7 @@ func TestRemoveSnapshot_LocalOnly_PreservesRemote(t *testing.T) {
|
|||||||
addBlob(t, store, blobA)
|
addBlob(t, store, blobA)
|
||||||
|
|
||||||
tv := vaultik.NewForTesting(store)
|
tv := vaultik.NewForTesting(store)
|
||||||
|
tv.Repositories = newLocalIndex(t, "snapshot-001")
|
||||||
|
|
||||||
opts := &vaultik.RemoveOptions{Force: true, LocalOnly: true}
|
opts := &vaultik.RemoveOptions{Force: true, LocalOnly: true}
|
||||||
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
||||||
@@ -262,6 +304,7 @@ func TestRemoveSnapshot_DefaultRemovesMetadataNotBlobs(t *testing.T) {
|
|||||||
addBlob(t, store, blobShared)
|
addBlob(t, store, blobShared)
|
||||||
|
|
||||||
tv := vaultik.NewForTesting(store)
|
tv := vaultik.NewForTesting(store)
|
||||||
|
tv.Repositories = newLocalIndex(t, "snapshot-001", "snapshot-002")
|
||||||
|
|
||||||
opts := &vaultik.RemoveOptions{Force: true}
|
opts := &vaultik.RemoveOptions{Force: true}
|
||||||
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
||||||
@@ -310,6 +353,84 @@ func TestRemoveSnapshot_DryRun(t *testing.T) {
|
|||||||
assert.Contains(t, tv.Stdout.String(), "[Dry run - no changes made]")
|
assert.Contains(t, tv.Stdout.String(), "[Dry run - no changes made]")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRemoveSnapshot_UnknownIDIsError checks that an ID naming no
|
||||||
|
// snapshot in the local index or on the destination store is an error
|
||||||
|
// under every option, and that nothing is reported as removed.
|
||||||
|
func TestRemoveSnapshot_UnknownIDIsError(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for name, opts := range map[string]vaultik.RemoveOptions{
|
||||||
|
"force": {Force: true},
|
||||||
|
"json": {JSON: true},
|
||||||
|
"dry-run": {Force: true, DryRun: true},
|
||||||
|
"local-only": {Force: true, LocalOnly: true},
|
||||||
|
} {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
store := newTestStorer()
|
||||||
|
addManifest(t, store, "snapshot-001", nil)
|
||||||
|
|
||||||
|
tv := vaultik.NewForTesting(store)
|
||||||
|
tv.Repositories = newLocalIndex(t, "snapshot-001")
|
||||||
|
|
||||||
|
result, err := tv.RemoveSnapshot("snapshot-999", &opts)
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "snapshot not found")
|
||||||
|
assert.False(t, result.RemoteRemoved)
|
||||||
|
assert.Empty(t, tv.Stdout.String())
|
||||||
|
assert.True(t, store.hasKey(remoteKeyPath("snapshot-001")))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRemoveSnapshot_NoMetadataOnDestination checks that removing a
|
||||||
|
// snapshot whose metadata is not on the destination store does not
|
||||||
|
// report metadata removed from it.
|
||||||
|
func TestRemoveSnapshot_NoMetadataOnDestination(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tv := vaultik.NewForTesting(newTestStorer())
|
||||||
|
tv.Repositories = newLocalIndex(t, "snapshot-001")
|
||||||
|
|
||||||
|
result, err := tv.RemoveSnapshot("snapshot-001",
|
||||||
|
&vaultik.RemoveOptions{Force: true})
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.False(t, result.RemoteRemoved)
|
||||||
|
|
||||||
|
out := tv.Stdout.String()
|
||||||
|
assert.Contains(t, out, "Removed snapshot 'snapshot-001' from local database")
|
||||||
|
assert.NotContains(t, out, "Removed snapshot metadata from remote storage")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRemoveSnapshot_NotInLocalIndex checks that removing a snapshot
|
||||||
|
// only the destination store holds removes its metadata there without
|
||||||
|
// reporting a removal from the local database.
|
||||||
|
func TestRemoveSnapshot_NotInLocalIndex(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
store := newTestStorer()
|
||||||
|
addManifest(t, store, "snapshot-001", nil)
|
||||||
|
|
||||||
|
tv := vaultik.NewForTesting(store)
|
||||||
|
tv.Repositories = newLocalIndex(t)
|
||||||
|
|
||||||
|
result, err := tv.RemoveSnapshot("snapshot-001",
|
||||||
|
&vaultik.RemoveOptions{Force: true})
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.True(t, result.RemoteRemoved)
|
||||||
|
assert.False(t, store.hasKey(remoteKeyPath("snapshot-001")))
|
||||||
|
|
||||||
|
out := tv.Stdout.String()
|
||||||
|
assert.NotContains(t, out, "from local database")
|
||||||
|
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
||||||
|
}
|
||||||
|
|
||||||
func TestRemoveAllSnapshots_RequiresForce(t *testing.T) {
|
func TestRemoveAllSnapshots_RequiresForce(t *testing.T) {
|
||||||
log.Initialize(log.Config{})
|
log.Initialize(log.Config{})
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
@@ -355,7 +476,8 @@ func TestRemoveAllSnapshots_WithForce(t *testing.T) {
|
|||||||
out := tv.Stdout.String()
|
out := tv.Stdout.String()
|
||||||
assert.Contains(t, out, "Removed 2 snapshot(s)")
|
assert.Contains(t, out, "Removed 2 snapshot(s)")
|
||||||
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
assert.Contains(t, out, "Removed snapshot metadata from remote storage")
|
||||||
assert.Contains(t, out, "vaultik prune")
|
// Remote nuke, the only caller, deletes the blobs next itself.
|
||||||
|
assert.NotContains(t, out, "vaultik prune")
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRemoveAllSnapshots_DryRun(t *testing.T) {
|
func TestRemoveAllSnapshots_DryRun(t *testing.T) {
|
||||||
|
|||||||
+106
-49
@@ -23,6 +23,8 @@ var (
|
|||||||
errNoSnapshotsInConfig = errors.New("no snapshots configured")
|
errNoSnapshotsInConfig = errors.New("no snapshots configured")
|
||||||
errSnapshotVerifyFailed = errors.New("verification failed")
|
errSnapshotVerifyFailed = errors.New("verification failed")
|
||||||
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
||||||
|
errSnapshotNotFound = errors.New("snapshot not found")
|
||||||
|
errSnapshotRemoveFailed = errors.New("snapshot(s) could not be removed")
|
||||||
errInvalidTableName = errors.New("invalid table name")
|
errInvalidTableName = errors.New("invalid table name")
|
||||||
errNoAgeRecipients = errors.New(
|
errNoAgeRecipients = errors.New(
|
||||||
"creating a snapshot needs at least one public key in " +
|
"creating a snapshot needs at least one public key in " +
|
||||||
@@ -643,7 +645,7 @@ func (v *Vaultik) confirmAndExecutePurge(
|
|||||||
"snapshot_id", snapshotID, "error", err)
|
"snapshot_id", snapshotID, "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("deleting snapshot %s from remote: %w", snapshotID, err)
|
return fmt.Errorf("deleting snapshot %s from remote: %w", snapshotID, err)
|
||||||
}
|
}
|
||||||
@@ -1153,6 +1155,11 @@ func (v *Vaultik) RemoveSnapshot(
|
|||||||
return result, err
|
return result, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
inLocalIndex, err := v.findSnapshotToRemove(snapshotID, opts)
|
||||||
|
if err != nil {
|
||||||
|
return result, err
|
||||||
|
}
|
||||||
|
|
||||||
if opts.DryRun {
|
if opts.DryRun {
|
||||||
return v.removeSnapshotDryRun(result, snapshotID, opts)
|
return v.removeSnapshotDryRun(result, snapshotID, opts)
|
||||||
}
|
}
|
||||||
@@ -1161,12 +1168,14 @@ func (v *Vaultik) RemoveSnapshot(
|
|||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if inLocalIndex {
|
||||||
log.Info("Removing snapshot from local database", "snapshot_id", snapshotID)
|
log.Info("Removing snapshot from local database", "snapshot_id", snapshotID)
|
||||||
|
|
||||||
err = v.deleteSnapshotFromLocalDB(snapshotID)
|
err = v.deleteSnapshotFromLocalDB(snapshotID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return result, fmt.Errorf("removing from local database: %w", err)
|
return result, fmt.Errorf("removing from local database: %w", err)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if !opts.LocalOnly {
|
if !opts.LocalOnly {
|
||||||
result.RemoteRemoved = v.removeSnapshotRemote(snapshotID, opts)
|
result.RemoteRemoved = v.removeSnapshotRemote(snapshotID, opts)
|
||||||
@@ -1184,9 +1193,11 @@ func (v *Vaultik) RemoveSnapshot(
|
|||||||
return result, v.outputRemoveJSON(result)
|
return result, v.outputRemoveJSON(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if inLocalIndex {
|
||||||
v.stdoutf("Removed snapshot '%s' from local database\n", snapshotID)
|
v.stdoutf("Removed snapshot '%s' from local database\n", snapshotID)
|
||||||
|
}
|
||||||
|
|
||||||
if !opts.LocalOnly && result.RemoteRemoved {
|
if result.RemoteRemoved {
|
||||||
v.printlnStdout("Removed snapshot metadata from remote storage")
|
v.printlnStdout("Removed snapshot metadata from remote storage")
|
||||||
v.stdoutf("\nNote: The removed snapshot's blobs remain on the remote. "+
|
v.stdoutf("\nNote: The removed snapshot's blobs remain on the remote. "+
|
||||||
"Run '%s' to delete any blobs no longer referenced by any "+
|
"Run '%s' to delete any blobs no longer referenced by any "+
|
||||||
@@ -1196,6 +1207,45 @@ func (v *Vaultik) RemoveSnapshot(
|
|||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// findSnapshotToRemove reports whether the local index has a row for
|
||||||
|
// snapshotID, and returns errSnapshotNotFound when there is nothing to
|
||||||
|
// remove: no row and, unless LocalOnly is set, no metadata on the
|
||||||
|
// destination store either. The destination store is consulted only
|
||||||
|
// when the local index has no row, so a snapshot in the local index can
|
||||||
|
// still be removed from it while the destination store is unreachable.
|
||||||
|
func (v *Vaultik) findSnapshotToRemove(
|
||||||
|
snapshotID string, opts *RemoveOptions,
|
||||||
|
) (bool, error) {
|
||||||
|
if v.Repositories != nil {
|
||||||
|
snap, err := v.Repositories.Snapshots.GetByID(v.ctx, snapshotID)
|
||||||
|
if err != nil {
|
||||||
|
return false, fmt.Errorf("looking up snapshot in local index: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if snap != nil {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if opts.LocalOnly {
|
||||||
|
return false, fmt.Errorf("%w in the local index: %s",
|
||||||
|
errSnapshotNotFound, snapshotID)
|
||||||
|
}
|
||||||
|
|
||||||
|
objects, err := v.remoteSnapshotObjects(snapshot.RemoteSnapshotKey(snapshotID))
|
||||||
|
if err != nil {
|
||||||
|
return false, fmt.Errorf("looking up snapshot on destination store: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(objects) == 0 {
|
||||||
|
return false, fmt.Errorf(
|
||||||
|
"%w in the local index or on the destination store: %s",
|
||||||
|
errSnapshotNotFound, snapshotID)
|
||||||
|
}
|
||||||
|
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
// removeSnapshotDryRun reports what RemoveSnapshot would do without
|
// removeSnapshotDryRun reports what RemoveSnapshot would do without
|
||||||
// making changes.
|
// making changes.
|
||||||
func (v *Vaultik) removeSnapshotDryRun(
|
func (v *Vaultik) removeSnapshotDryRun(
|
||||||
@@ -1255,14 +1305,14 @@ func (v *Vaultik) confirmRemoveSnapshot(snapshotID string, opts *RemoveOptions)
|
|||||||
// didn't finish and to run `vaultik snapshot remove` for the snapshot
|
// didn't finish and to run `vaultik snapshot remove` for the snapshot
|
||||||
// again once the destination store is reachable (`vaultik prune` never
|
// again once the destination store is reachable (`vaultik prune` never
|
||||||
// removes snapshot metadata). Returns true when the remote removal
|
// removes snapshot metadata). Returns true when the remote removal
|
||||||
// succeeded.
|
// succeeded and deleted something.
|
||||||
func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) bool {
|
func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) bool {
|
||||||
log.Info("Removing snapshot metadata from remote storage",
|
log.Info("Removing snapshot metadata from remote storage",
|
||||||
"snapshot_id", snapshotID)
|
"snapshot_id", snapshotID)
|
||||||
|
|
||||||
remoteKey := snapshot.RemoteSnapshotKey(snapshotID)
|
remoteKey := snapshot.RemoteSnapshotKey(snapshotID)
|
||||||
|
|
||||||
err := v.deleteRemoteSnapshotByKey(remoteKey)
|
deleted, err := v.deleteRemoteSnapshotByKey(remoteKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("Could not remove snapshot metadata from remote storage; "+
|
log.Warn("Could not remove snapshot metadata from remote storage; "+
|
||||||
"run '"+snapshotRemoveCommandHint+"' with the snapshot's ID "+
|
"run '"+snapshotRemoveCommandHint+"' with the snapshot's ID "+
|
||||||
@@ -1280,7 +1330,7 @@ func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) b
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
return true
|
return deleted > 0
|
||||||
}
|
}
|
||||||
|
|
||||||
// RemoveAllSnapshots removes every snapshot known to the local
|
// RemoveAllSnapshots removes every snapshot known to the local
|
||||||
@@ -1464,8 +1514,8 @@ func (v *Vaultik) handleRemoveAllDryRun(
|
|||||||
|
|
||||||
// executeRemoveAll deletes every local snapshot and, unless LocalOnly
|
// executeRemoveAll deletes every local snapshot and, unless LocalOnly
|
||||||
// is set, every corresponding remote metadata directory plus any
|
// is set, every corresponding remote metadata directory plus any
|
||||||
// orphan remote keys. Blobs are not touched; the printed prune-command
|
// orphan remote keys. Blobs are not touched. It returns an error when
|
||||||
// hint is the next step.
|
// any snapshot could not be removed.
|
||||||
func (v *Vaultik) executeRemoveAll(
|
func (v *Vaultik) executeRemoveAll(
|
||||||
localSnaps, orphanRemoteKeys []string, opts *RemoveOptions,
|
localSnaps, orphanRemoteKeys []string, opts *RemoveOptions,
|
||||||
) (*RemoveResult, error) {
|
) (*RemoveResult, error) {
|
||||||
@@ -1479,18 +1529,11 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
|
|
||||||
result := &RemoveResult{}
|
result := &RemoveResult{}
|
||||||
|
|
||||||
remoteErrors := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
failures := v.removeAllLocalSnapshots(localSnaps, opts, result)
|
||||||
|
|
||||||
if !opts.LocalOnly {
|
if !opts.LocalOnly {
|
||||||
remoteErrors += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
failures += v.removeAllOrphanRemotes(orphanRemoteKeys, result)
|
||||||
|
result.RemoteRemoved = failures == 0
|
||||||
if remoteErrors == 0 {
|
|
||||||
result.RemoteRemoved = true
|
|
||||||
} else if v.UI != nil {
|
|
||||||
v.UI.Warningf("Some remote metadata deletions failed. Run '%s' "+
|
|
||||||
"once the remote is healthy to clean up unreferenced blobs.",
|
|
||||||
pruneCommandHint)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if v.SnapshotManager != nil {
|
if v.SnapshotManager != nil {
|
||||||
@@ -1500,6 +1543,10 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if failures > 0 {
|
||||||
|
return result, fmt.Errorf("%d %w", failures, errSnapshotRemoveFailed)
|
||||||
|
}
|
||||||
|
|
||||||
if opts.JSON {
|
if opts.JSON {
|
||||||
return result, v.outputRemoveJSON(result)
|
return result, v.outputRemoveJSON(result)
|
||||||
}
|
}
|
||||||
@@ -1508,9 +1555,6 @@ func (v *Vaultik) executeRemoveAll(
|
|||||||
|
|
||||||
if !opts.LocalOnly && result.RemoteRemoved {
|
if !opts.LocalOnly && result.RemoteRemoved {
|
||||||
v.printlnStdout("Removed snapshot metadata from remote storage")
|
v.printlnStdout("Removed snapshot metadata from remote storage")
|
||||||
v.stdoutf("\nNote: Removed snapshots' blobs remain on the remote. "+
|
|
||||||
"Run '%s' to delete any blobs no longer referenced by any "+
|
|
||||||
"remaining remote snapshot.\n", pruneCommandHint)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return result, nil
|
return result, nil
|
||||||
@@ -1553,12 +1597,13 @@ func (v *Vaultik) deleteIncompleteSnapshot(snapshotIDStr string, id any) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// removeAllLocalSnapshots removes each known local snapshot (and, unless
|
// removeAllLocalSnapshots removes each known local snapshot (and, unless
|
||||||
// --local-only, its remote metadata), returning the number of remote
|
// --local-only, its remote metadata), returning the number of snapshots
|
||||||
// deletion failures.
|
// it could not remove. A snapshot left in the local index keeps its
|
||||||
|
// remote metadata too, since it is not an orphan remote key.
|
||||||
func (v *Vaultik) removeAllLocalSnapshots(
|
func (v *Vaultik) removeAllLocalSnapshots(
|
||||||
localSnaps []string, opts *RemoveOptions, result *RemoveResult,
|
localSnaps []string, opts *RemoveOptions, result *RemoveResult,
|
||||||
) int {
|
) int {
|
||||||
remoteErrors := 0
|
failures := 0
|
||||||
|
|
||||||
for _, snapshotID := range localSnaps {
|
for _, snapshotID := range localSnaps {
|
||||||
log.Info("Removing snapshot", "snapshot_id", snapshotID)
|
log.Info("Removing snapshot", "snapshot_id", snapshotID)
|
||||||
@@ -1568,24 +1613,26 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
|||||||
log.Error("Failed to remove from local database",
|
log.Error("Failed to remove from local database",
|
||||||
"snapshot_id", snapshotID, "error", err)
|
"snapshot_id", snapshotID, "error", err)
|
||||||
|
|
||||||
|
failures++
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
if !opts.LocalOnly {
|
if !opts.LocalOnly {
|
||||||
err = v.deleteRemoteSnapshotByKey(
|
_, err = v.deleteRemoteSnapshotByKey(
|
||||||
snapshot.RemoteSnapshotKey(snapshotID))
|
snapshot.RemoteSnapshotKey(snapshotID))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("Failed to remove snapshot metadata from remote",
|
log.Warn("Failed to remove snapshot metadata from remote",
|
||||||
"snapshot_id", snapshotID, "error", err)
|
"snapshot_id", snapshotID, "error", err)
|
||||||
|
|
||||||
remoteErrors++
|
failures++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
result.SnapshotsRemoved = append(result.SnapshotsRemoved, snapshotID)
|
result.SnapshotsRemoved = append(result.SnapshotsRemoved, snapshotID)
|
||||||
}
|
}
|
||||||
|
|
||||||
return remoteErrors
|
return failures
|
||||||
}
|
}
|
||||||
|
|
||||||
// removeAllOrphanRemotes deletes remote snapshot metadata that has no
|
// removeAllOrphanRemotes deletes remote snapshot metadata that has no
|
||||||
@@ -1598,7 +1645,7 @@ func (v *Vaultik) removeAllOrphanRemotes(
|
|||||||
for _, key := range orphanRemoteKeys {
|
for _, key := range orphanRemoteKeys {
|
||||||
log.Info("Removing orphan remote snapshot", "remote_key", key)
|
log.Info("Removing orphan remote snapshot", "remote_key", key)
|
||||||
|
|
||||||
err := v.deleteRemoteSnapshotByKey(key)
|
_, err := v.deleteRemoteSnapshotByKey(key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("Failed to remove orphan from remote",
|
log.Warn("Failed to remove orphan from remote",
|
||||||
"remote_key", key, "error", err)
|
"remote_key", key, "error", err)
|
||||||
@@ -1645,34 +1692,44 @@ func (v *Vaultik) deleteSnapshotFromLocalDB(snapshotID string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// deleteRemoteSnapshotByKey removes everything under
|
// deleteRemoteSnapshotByKey removes everything under
|
||||||
// metadata/<remoteKey>/ on the destination store. The argument is a
|
// metadata/<remoteKey>/ on the destination store and returns how many
|
||||||
// remote key (double-SHA256 derivation), not a human snapshot ID;
|
// objects it deleted. The argument is a remote key (double-SHA256
|
||||||
// callers that have a human ID must hash via snapshot.RemoteSnapshotKey
|
// derivation), not a human snapshot ID; callers that have a human ID
|
||||||
// first.
|
// must hash via snapshot.RemoteSnapshotKey first.
|
||||||
func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) error {
|
func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) (int, error) {
|
||||||
prefix := fmt.Sprintf("metadata/%s/", remoteKey)
|
objectsToDelete, err := v.remoteSnapshotObjects(remoteKey)
|
||||||
objectCh := v.Storage.ListStream(v.ctx, prefix)
|
|
||||||
|
|
||||||
var objectsToDelete []string
|
|
||||||
|
|
||||||
for object := range objectCh {
|
|
||||||
if object.Err != nil {
|
|
||||||
return fmt.Errorf("listing objects: %w", object.Err)
|
|
||||||
}
|
|
||||||
|
|
||||||
objectsToDelete = append(objectsToDelete, object.Key)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, key := range objectsToDelete {
|
|
||||||
err := v.Storage.Delete(v.ctx, key)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("removing %s: %w", key, err)
|
return 0, err
|
||||||
|
}
|
||||||
|
|
||||||
|
for i, key := range objectsToDelete {
|
||||||
|
err = v.Storage.Delete(v.ctx, key)
|
||||||
|
if err != nil {
|
||||||
|
return i, fmt.Errorf("removing %s: %w", key, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Debug("Deleted remote object", "key", key)
|
log.Debug("Deleted remote object", "key", key)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return len(objectsToDelete), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// remoteSnapshotObjects lists the keys of every object under
|
||||||
|
// metadata/<remoteKey>/ on the destination store.
|
||||||
|
func (v *Vaultik) remoteSnapshotObjects(remoteKey string) ([]string, error) {
|
||||||
|
prefix := fmt.Sprintf("metadata/%s/", remoteKey)
|
||||||
|
|
||||||
|
var keys []string
|
||||||
|
|
||||||
|
for object := range v.Storage.ListStream(v.ctx, prefix) {
|
||||||
|
if object.Err != nil {
|
||||||
|
return nil, fmt.Errorf("listing objects: %w", object.Err)
|
||||||
|
}
|
||||||
|
|
||||||
|
keys = append(keys, object.Key)
|
||||||
|
}
|
||||||
|
|
||||||
|
return keys, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// outputRemoveJSON outputs the removal result as JSON
|
// outputRemoveJSON outputs the removal result as JSON
|
||||||
|
|||||||
@@ -0,0 +1,75 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/config"
|
||||||
|
"sneak.berlin/go/vaultik/internal/database"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// --skip-errors skips only a file that cannot be opened or read. A local
|
||||||
|
// index error while a directory is recorded stops the backup, and the
|
||||||
|
// snapshot is not recorded as complete. See
|
||||||
|
// https://git.eeqj.de/sneak/vaultik/issues/284.
|
||||||
|
func TestSkipErrorsBackupStopsOnIndexErrorRecordingDirectory(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
// The scan walks the source path with symlinks resolved, so dirPath
|
||||||
|
// must be spelled the same way to match the row the scan inserts.
|
||||||
|
tempDir, err := filepath.EvalSymlinks(t.TempDir())
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
srcDir := filepath.Join(tempDir, "src")
|
||||||
|
dirPath := filepath.Join(srcDir, "dir")
|
||||||
|
|
||||||
|
fs := afero.NewOsFs()
|
||||||
|
require.NoError(t, fs.MkdirAll(dirPath, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dirPath, "file.txt"), []byte("file content"), 0o644))
|
||||||
|
|
||||||
|
cfg := faultTestConfig()
|
||||||
|
cfg.IndexPath = filepath.Join(tempDir, "index.sqlite")
|
||||||
|
cfg.Snapshots = map[string]config.SnapshotConfig{
|
||||||
|
"tree": {Paths: []string{srcDir}},
|
||||||
|
}
|
||||||
|
|
||||||
|
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, err := database.New(ctx, cfg.IndexPath)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
// The local index refuses the directory's files row.
|
||||||
|
_, err = db.Conn().ExecContext(ctx, fmt.Sprintf(`
|
||||||
|
CREATE TRIGGER refuse_directory BEFORE INSERT ON files
|
||||||
|
WHEN NEW.path = '%s'
|
||||||
|
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`, dirPath))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
repos := database.NewRepositories(db)
|
||||||
|
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
|
||||||
|
|
||||||
|
err = v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
|
||||||
|
SkipErrors: true,
|
||||||
|
Snapshots: []string{"tree"},
|
||||||
|
})
|
||||||
|
require.ErrorContains(t, err, "simulated index error")
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Len(t, snapshots, 1)
|
||||||
|
assert.Nil(t, snapshots[0].CompletedAt)
|
||||||
|
}
|
||||||
@@ -9,7 +9,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// errStorageBindingMismatch is returned when the local index database is
|
// errStorageBindingMismatch is returned when the local index database is
|
||||||
// bound to a different destination than the configured storage_url.
|
// bound to a different destination than the configured one.
|
||||||
var errStorageBindingMismatch = errors.New(
|
var errStorageBindingMismatch = errors.New(
|
||||||
"local index is bound to a different backup destination")
|
"local index is bound to a different backup destination")
|
||||||
|
|
||||||
@@ -24,8 +24,9 @@ var errStorageBindingMismatch = errors.New(
|
|||||||
//
|
//
|
||||||
// Behaviour:
|
// Behaviour:
|
||||||
// - On first use (empty stored value), record the configured
|
// - On first use (empty stored value), record the configured
|
||||||
// storage_url and log the binding.
|
// destination (storage_url, or the s3.* endpoint, bucket and
|
||||||
// - When the stored value matches the configured storage_url, do
|
// prefix) and log the binding.
|
||||||
|
// - When the stored value matches the configured destination, do
|
||||||
// nothing and return nil.
|
// nothing and return nil.
|
||||||
// - When the two differ, refuse with an error that tells the user
|
// - When the two differ, refuse with an error that tells the user
|
||||||
// how to recover (revert the config, or run `vaultik database
|
// how to recover (revert the config, or run `vaultik database
|
||||||
@@ -46,12 +47,10 @@ func (v *Vaultik) EnsureStorageBinding() error {
|
|||||||
|
|
||||||
configured := v.Config.StorageURL
|
configured := v.Config.StorageURL
|
||||||
if configured == "" {
|
if configured == "" {
|
||||||
// Some legacy configs still use the split s3.* keys instead of
|
// The s3.* form: bind on its endpoint, bucket and prefix,
|
||||||
// storage_url. Falling back to a synthetic URL for those would
|
// written the way storage_url would name the same destination.
|
||||||
// hide the fact that the binding is loose. Instead, treat
|
s3 := v.Config.S3
|
||||||
// unset as "nothing to bind against" — the check is
|
configured = "s3://" + s3.Bucket + "/" + s3.Prefix + "?endpoint=" + s3.Endpoint
|
||||||
// necessarily best-effort for those configs.
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
||||||
@@ -96,7 +95,8 @@ func buildBindingMismatchMessage(stored, configured string) string {
|
|||||||
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
||||||
"\n" +
|
"\n" +
|
||||||
"To proceed, either:\n" +
|
"To proceed, either:\n" +
|
||||||
" - revert storage_url in your config to the bound destination, or\n" +
|
" - revert storage_url (or the s3.* endpoint, bucket and prefix) in your\n" +
|
||||||
|
" config to the bound destination, or\n" +
|
||||||
" - run 'vaultik database delete' to discard the local index and rebuild it\n" +
|
" - run 'vaultik database delete' to discard the local index and rebuild it\n" +
|
||||||
" from a fresh full backup against the new destination"
|
" from a fresh full backup against the new destination"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -76,3 +76,51 @@ func TestEnsureStorageBinding_MismatchRefuses(t *testing.T) {
|
|||||||
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
||||||
assert.Contains(t, err.Error(), "vaultik database delete")
|
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A config with no storage_url names its destination with s3.endpoint,
|
||||||
|
// s3.bucket and s3.prefix; a change to any of them must be refused just
|
||||||
|
// as a changed storage_url is.
|
||||||
|
func TestEnsureStorageBinding_S3FieldsChangeRefuses(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
bound := config.S3Config{
|
||||||
|
Endpoint: "https://s3.example.com",
|
||||||
|
Bucket: "bucket-a",
|
||||||
|
Prefix: "backups",
|
||||||
|
}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
changed config.S3Config
|
||||||
|
}{
|
||||||
|
{"bucket", config.S3Config{
|
||||||
|
Endpoint: bound.Endpoint, Bucket: "bucket-b", Prefix: bound.Prefix,
|
||||||
|
}},
|
||||||
|
{"prefix", config.S3Config{
|
||||||
|
Endpoint: bound.Endpoint, Bucket: bound.Bucket, Prefix: "other",
|
||||||
|
}},
|
||||||
|
{"endpoint", config.S3Config{
|
||||||
|
Endpoint: "https://s3.other.com", Bucket: bound.Bucket, Prefix: bound.Prefix,
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tv, _ := buildBindTestVaultik(t, "")
|
||||||
|
|
||||||
|
tv.Config.S3 = bound
|
||||||
|
require.NoError(t, tv.EnsureStorageBinding(), "first use binds")
|
||||||
|
require.NoError(t, tv.EnsureStorageBinding(), "same destination passes")
|
||||||
|
|
||||||
|
tv.Config.S3 = tt.changed
|
||||||
|
err := tv.EnsureStorageBinding()
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.Contains(t, err.Error(),
|
||||||
|
"local index is bound to a different backup destination")
|
||||||
|
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user