Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3587f4ae94 |
@@ -349,11 +349,9 @@ the backup destination store. Blobs are NOT touched — deleting blobs
|
||||
requires reading every remaining remote manifest (the destination store
|
||||
may hold snapshots this host doesn't know about), which is what
|
||||
`vaultik prune` does. On success the command 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; with
|
||||
`--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
|
||||
prune` invocation to run as a follow-up. 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
|
||||
remove <snapshot-id>` again once the store is reachable to remove the
|
||||
snapshot's metadata from it (`vaultik prune` does not). To wipe everything
|
||||
@@ -414,8 +412,7 @@ destination store.
|
||||
entirely. Remote storage is unaffected; the next backup will do a full
|
||||
scan and re-deduplicate against existing remote blobs, and the local
|
||||
index will re-bind to the currently configured storage destination.
|
||||
Use this after changing `storage_url`, or `s3.endpoint`, `s3.bucket` or
|
||||
`s3.prefix`, to a different destination.
|
||||
Use this after changing `storage_url` to a different destination.
|
||||
* `--force`: Skip confirmation prompt
|
||||
|
||||
---
|
||||
|
||||
@@ -22,52 +22,6 @@ the tag exists and is exercised; what is left is merging `next` to
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-13: Made an S3 listing that the server cuts short an error
|
||||
([issue #297](https://git.eeqj.de/sneak/vaultik/issues/297)). A page
|
||||
marked truncated with no continuation token, as a server answering V2
|
||||
listings V1-style sends, ended the listing without an error, so
|
||||
`prune` saw only the snapshots on the first page and could delete
|
||||
blobs the rest still use. Such a page is now a listing error, and
|
||||
`prune`, `snapshot list` and `remote nuke` handle it as a destination
|
||||
that cannot be listed.
|
||||
|
||||
- 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
|
||||
@@ -75,15 +29,6 @@ the tag exists and is exercised; what is left is merging `next` to
|
||||
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
|
||||
([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
|
||||
|
||||
@@ -255,12 +255,8 @@ manifest (the destination store may hold snapshots this host doesn't
|
||||
know about), which is what 'vaultik prune' does. On success the command
|
||||
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
|
||||
snapshot locally without touching the destination store); the snapshot
|
||||
must then be in the local index.
|
||||
snapshot locally without touching the destination store).
|
||||
|
||||
If the remote is unreachable, the local-database removal still completes
|
||||
and a warning is emitted; run 'vaultik snapshot remove <snapshot-id>' again
|
||||
|
||||
+2
-29
@@ -18,13 +18,6 @@ import (
|
||||
"github.com/aws/smithy-go/logging"
|
||||
)
|
||||
|
||||
// ErrNoContinuationToken is returned by ListObjects and ListObjectsStream
|
||||
// when the server marks a page of a listing truncated but sends no
|
||||
// continuation token to fetch the rest with, as a server answering V2
|
||||
// listings V1-style does.
|
||||
var ErrNoContinuationToken = errors.New(
|
||||
"server marked the listing truncated but sent no continuation token")
|
||||
|
||||
// Client wraps the AWS S3 client for vaultik operations.
|
||||
// It provides a simplified interface for S3 operations with automatic
|
||||
// prefix handling and connection management. All operations are performed
|
||||
@@ -221,7 +214,7 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
page, err := paginator.NextPage(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -242,26 +235,6 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
// nextListPage fetches the next page of a listing. The SDK paginator takes
|
||||
// a page marked truncated without a continuation token for the last page
|
||||
// and stops without an error, so such a page is ErrNoContinuationToken
|
||||
// here instead.
|
||||
func nextListPage(
|
||||
ctx context.Context, paginator *s3.ListObjectsV2Paginator,
|
||||
) (*s3.ListObjectsV2Output, error) {
|
||||
page, err := paginator.NextPage(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
truncated := aws.ToBool(page.IsTruncated)
|
||||
if truncated && aws.ToString(page.NextContinuationToken) == "" {
|
||||
return nil, ErrNoContinuationToken
|
||||
}
|
||||
|
||||
return page, nil
|
||||
}
|
||||
|
||||
// HeadObject checks if an object exists in S3.
|
||||
// Returns true if the object exists, false otherwise.
|
||||
// The key is automatically prefixed with the configured prefix.
|
||||
@@ -327,7 +300,7 @@ func (c *Client) ListObjectsStream(
|
||||
})
|
||||
|
||||
for paginator.HasMorePages() {
|
||||
page, err := nextListPage(ctx, paginator)
|
||||
page, err := paginator.NextPage(ctx)
|
||||
if err != nil {
|
||||
ch <- ObjectInfo{Err: err}
|
||||
|
||||
|
||||
@@ -1312,13 +1312,6 @@ func (s *Scanner) processPhase(
|
||||
|
||||
// Process each file
|
||||
for _, fileToProcess := range filesToProcess {
|
||||
// Check context cancellation
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
}
|
||||
|
||||
// Update progress
|
||||
if s.progress != nil {
|
||||
s.progress.GetStats().CurrentFile.Store(fileToProcess.Path)
|
||||
@@ -1355,32 +1348,13 @@ func (s *Scanner) processPhase(
|
||||
return s.finalizeProcessPhase(ctx, result)
|
||||
}
|
||||
|
||||
// processFileWithErrorHandling records a directory or symlink, or wraps
|
||||
// processFileStreaming for a regular file with error recovery for deleted
|
||||
// files and skip-errors mode. Returns (skipped, error).
|
||||
// processFileWithErrorHandling wraps processFileStreaming with error recovery for
|
||||
// deleted files and skip-errors mode. Returns (skipped, error).
|
||||
func (s *Scanner) processFileWithErrorHandling(
|
||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||
) (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)
|
||||
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
|
||||
// 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
|
||||
@@ -1418,7 +1392,12 @@ func (s *Scanner) processFileWithErrorHandling(
|
||||
|
||||
// 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.
|
||||
// Phase 1 counts no directories, so a directory is not counted here.
|
||||
func countFailedFile(fileToProcess *FileToProcess, result *ScanResult) {
|
||||
if fileToProcess.FileInfo.IsDir() {
|
||||
return
|
||||
}
|
||||
|
||||
result.FilesFailed++
|
||||
result.BytesScanned -= fileToProcess.FileInfo.Size()
|
||||
}
|
||||
@@ -1791,11 +1770,16 @@ func (e *packerError) Error() string { return e.err.Error() }
|
||||
|
||||
func (e *packerError) Unwrap() error { return e.err }
|
||||
|
||||
// processFileStreaming processes a regular file by streaming chunks directly
|
||||
// to the packer
|
||||
// processFileStreaming processes a file by streaming chunks directly to the packer
|
||||
func (s *Scanner) processFileStreaming(
|
||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||
) 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)
|
||||
if err != nil {
|
||||
return fmt.Errorf("opening file: %w", wrapPermissionError(fileToProcess.Path, err))
|
||||
|
||||
@@ -82,32 +82,6 @@ func (f *readFailFs) Open(name string) (afero.File, error) {
|
||||
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
|
||||
// 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.
|
||||
@@ -154,16 +128,15 @@ func writeSkipErrorTestFile(t *testing.T, fs afero.Fs, path, content string) {
|
||||
}
|
||||
}
|
||||
|
||||
// runSkipErrorScan scans source on fs under ctx with the given skip-errors
|
||||
// setting, printing user-facing messages to uiw (nil discards them), and
|
||||
// returns the repositories (for inspection) and the scan error.
|
||||
// runSkipErrorScan scans source on fs with the given skip-errors setting,
|
||||
// printing user-facing messages to uiw (nil discards them), and returns the
|
||||
// repositories (for inspection) and the scan error.
|
||||
func runSkipErrorScan(
|
||||
ctx context.Context, t *testing.T, fs afero.Fs, source string,
|
||||
skipErrors bool, uiw *ui.Writer,
|
||||
t *testing.T, fs afero.Fs, source string, skipErrors bool, uiw *ui.Writer,
|
||||
) (*database.Repositories, error) {
|
||||
t.Helper()
|
||||
|
||||
db, err := database.New(ctx, ":memory:")
|
||||
db, err := database.NewTestDB()
|
||||
if err != nil {
|
||||
t.Fatalf("create test db: %v", err)
|
||||
}
|
||||
@@ -188,6 +161,7 @@ func runSkipErrorScan(
|
||||
SkipErrors: skipErrors,
|
||||
})
|
||||
|
||||
ctx := context.Background()
|
||||
snapshotID := "test-snapshot-skip-errors"
|
||||
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
||||
|
||||
@@ -211,7 +185,7 @@ func TestScannerPackingFailureAbortsUnderSkipErrors(t *testing.T) {
|
||||
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
||||
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
||||
|
||||
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected scan to abort on the packer error, got nil")
|
||||
}
|
||||
@@ -238,7 +212,7 @@ func TestScannerReadErrorAbortsWithoutSkipErrors(t *testing.T) {
|
||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||
|
||||
_, err := runSkipErrorScan(context.Background(), t, fs, "/source", false, nil)
|
||||
_, err := runSkipErrorScan(t, fs, "/source", false, nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected scan to fail on the read error, got nil")
|
||||
}
|
||||
@@ -254,7 +228,7 @@ func TestScannerReadErrorSkippedWithSkipErrors(t *testing.T) {
|
||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||
|
||||
repos, err := runSkipErrorScan(context.Background(), t, fs, "/source", true, nil)
|
||||
repos, err := runSkipErrorScan(t, fs, "/source", true, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||
}
|
||||
@@ -293,7 +267,7 @@ func TestScannerUnreadableSymlinkAbortsWithoutSkipErrors(t *testing.T) {
|
||||
sourceDir, linkPath := writeSymlinkSource(t)
|
||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||
|
||||
_, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, false, nil)
|
||||
_, err := runSkipErrorScan(t, fs, sourceDir, false, nil)
|
||||
if !errors.Is(err, os.ErrNotExist) {
|
||||
t.Fatalf("expected scan to fail on the removed symlink, got %v", err)
|
||||
}
|
||||
@@ -309,7 +283,7 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
||||
fs := &linkRemovedAfterLstatFs{t: t, target: linkPath}
|
||||
uiw := ui.NewWithColor(io.Discard, false)
|
||||
|
||||
repos, err := runSkipErrorScan(context.Background(), t, fs, sourceDir, true, uiw)
|
||||
repos, err := runSkipErrorScan(t, fs, sourceDir, true, uiw)
|
||||
if err != nil {
|
||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||
}
|
||||
@@ -328,56 +302,3 @@ func TestScannerUnreadableSymlinkSkippedWithSkipErrors(t *testing.T) {
|
||||
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())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
@@ -23,16 +22,13 @@ import (
|
||||
// s3TestBucket is the bucket created for each in-process S3 server.
|
||||
const s3TestBucket = "test-bucket"
|
||||
|
||||
// Credentials for the tests' S3 clients. The test servers accept any.
|
||||
// Credentials for the tests that build a storer from a config.Config. The
|
||||
// in-process S3 server accepts any.
|
||||
const (
|
||||
s3TestAccessKeyID = "key"
|
||||
s3TestSecretAccessKey = "secret"
|
||||
)
|
||||
|
||||
// s3TestRegion is the region of the S3 clients built here with
|
||||
// s3.NewClient. The test servers ignore it.
|
||||
const s3TestRegion = "us-east-1"
|
||||
|
||||
// newS3Storer builds an s3:// backend backed by a fresh in-process
|
||||
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
|
||||
// over httptest) that internal/s3 and the not-found regression test use,
|
||||
@@ -59,7 +55,7 @@ func newS3Storer(t *testing.T) storage.Storer {
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: "test",
|
||||
SecretAccessKey: "test",
|
||||
Region: s3TestRegion,
|
||||
Region: "us-east-1",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
@@ -97,68 +93,6 @@ func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3ListingCutShortIsAnError checks that List and ListStream fail on a
|
||||
// listing page marked truncated that carries no continuation token, as a
|
||||
// server answering V2 listings V1-style sends it. Taken as the last page,
|
||||
// it would hide every later key, and prune would treat the blobs of the
|
||||
// snapshots listed there as unreferenced.
|
||||
func TestS3ListingCutShortIsAnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
prefix = "metadata/"
|
||||
truncatedPage = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<Name>test-bucket</Name>
|
||||
<Prefix>metadata/</Prefix>
|
||||
<KeyCount>1</KeyCount>
|
||||
<MaxKeys>1</MaxKeys>
|
||||
<IsTruncated>true</IsTruncated>
|
||||
<Contents>
|
||||
<Key>metadata/snap/manifest.json.zst</Key>
|
||||
<Size>8</Size>
|
||||
</Contents>
|
||||
</ListBucketResult>`
|
||||
)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = io.WriteString(w, truncatedPage)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
client, err := s3.NewClient(context.Background(), s3.Config{
|
||||
Endpoint: srv.URL,
|
||||
Bucket: s3TestBucket,
|
||||
AccessKeyID: s3TestAccessKeyID,
|
||||
SecretAccessKey: s3TestSecretAccessKey,
|
||||
Region: s3TestRegion,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("new client: %v", err)
|
||||
}
|
||||
|
||||
storer := storage.NewS3Storer(client)
|
||||
ctx := context.Background()
|
||||
|
||||
keys, err := storer.List(ctx, prefix)
|
||||
if !errors.Is(err, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("List = %q, %v; want ErrNoContinuationToken", keys, err)
|
||||
}
|
||||
|
||||
var streamErr error
|
||||
|
||||
for obj := range storer.ListStream(ctx, prefix) {
|
||||
if obj.Err != nil {
|
||||
streamErr = obj.Err
|
||||
}
|
||||
}
|
||||
|
||||
if !errors.Is(streamErr, s3.ErrNoContinuationToken) {
|
||||
t.Errorf("ListStream error = %v, want ErrNoContinuationToken", streamErr)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
||||
// writes: the README's remote storage layout, with the prefix joined to
|
||||
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
||||
|
||||
@@ -3,7 +3,6 @@ package vaultik
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -48,10 +47,6 @@ const (
|
||||
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
|
||||
// timestamp.
|
||||
const minSnapshotIDParts = 2
|
||||
@@ -150,10 +145,10 @@ func parseSnapshotName(snapshotID, hostname string) string {
|
||||
// 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
|
||||
// duration units. Following Go, m is minutes and mo is months. A bare number,
|
||||
// an unknown unit, a negative value, and a total longer than maxDuration are
|
||||
// all rejected. Outside the Go units the input must be whole numbers each
|
||||
// followed directly by a unit (2w3d), so 1.0y and 30 days are errors;
|
||||
// decimals work only in Go units (1.5h).
|
||||
// an unknown unit, and a negative value are all rejected. Outside the Go
|
||||
// units the input must be whole numbers each followed directly by a unit
|
||||
// (2w3d), so 1.0y and 30 days are errors; decimals work only in Go units
|
||||
// (1.5h).
|
||||
func parseDuration(s string) (time.Duration, error) {
|
||||
if strings.HasPrefix(strings.TrimSpace(s), "-") {
|
||||
return 0, errNegativeDuration
|
||||
@@ -186,30 +181,19 @@ func parseDuration(s string) (time.Duration, error) {
|
||||
return 0, fmt.Errorf("invalid number %q: %w", match[1], err)
|
||||
}
|
||||
|
||||
var unitLength time.Duration
|
||||
|
||||
unit := strings.ToLower(match[2])
|
||||
switch unit {
|
||||
case "d", "day", "days":
|
||||
unitLength = day
|
||||
total += time.Duration(n) * day
|
||||
case "w", "week", "weeks":
|
||||
unitLength = week
|
||||
total += time.Duration(n) * week
|
||||
case "mo", "month", "months":
|
||||
unitLength = month
|
||||
total += time.Duration(n) * month
|
||||
case "y", "year", "years":
|
||||
unitLength = year
|
||||
total += time.Duration(n) * year
|
||||
default:
|
||||
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
|
||||
|
||||
@@ -111,17 +111,6 @@ func TestParseDuration(t *testing.T) {
|
||||
{"30 days", 0, true},
|
||||
{"30 days ago", 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 {
|
||||
|
||||
@@ -267,29 +267,6 @@ func TestPurgeOlderThan_WithNameFilter(t *testing.T) {
|
||||
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) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
@@ -3,21 +3,17 @@ package vaultik_test
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"io"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/klauspost/compress/zstd"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
"sneak.berlin/go/vaultik/internal/types"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
@@ -218,37 +214,6 @@ func addBlob(t *testing.T, store *testStorer, hash string) {
|
||||
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
|
||||
// ============================================================================
|
||||
@@ -268,7 +233,6 @@ func TestRemoveSnapshot_LocalOnly_PreservesRemote(t *testing.T) {
|
||||
addBlob(t, store, blobA)
|
||||
|
||||
tv := vaultik.NewForTesting(store)
|
||||
tv.Repositories = newLocalIndex(t, "snapshot-001")
|
||||
|
||||
opts := &vaultik.RemoveOptions{Force: true, LocalOnly: true}
|
||||
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
||||
@@ -304,7 +268,6 @@ func TestRemoveSnapshot_DefaultRemovesMetadataNotBlobs(t *testing.T) {
|
||||
addBlob(t, store, blobShared)
|
||||
|
||||
tv := vaultik.NewForTesting(store)
|
||||
tv.Repositories = newLocalIndex(t, "snapshot-001", "snapshot-002")
|
||||
|
||||
opts := &vaultik.RemoveOptions{Force: true}
|
||||
result, err := tv.RemoveSnapshot("snapshot-001", opts)
|
||||
@@ -353,84 +316,6 @@ func TestRemoveSnapshot_DryRun(t *testing.T) {
|
||||
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) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
@@ -23,7 +23,6 @@ var (
|
||||
errNoSnapshotsInConfig = errors.New("no snapshots configured")
|
||||
errSnapshotVerifyFailed = errors.New("verification failed")
|
||||
errRemoveAllNeedsForce = errors.New("--all requires --force")
|
||||
errSnapshotNotFound = errors.New("snapshot not found")
|
||||
errInvalidTableName = errors.New("invalid table name")
|
||||
errNoAgeRecipients = errors.New(
|
||||
"creating a snapshot needs at least one public key in " +
|
||||
@@ -644,7 +643,7 @@ func (v *Vaultik) confirmAndExecutePurge(
|
||||
"snapshot_id", snapshotID, "error", err)
|
||||
}
|
||||
|
||||
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||
err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
|
||||
if err != nil {
|
||||
return fmt.Errorf("deleting snapshot %s from remote: %w", snapshotID, err)
|
||||
}
|
||||
@@ -1154,11 +1153,6 @@ func (v *Vaultik) RemoveSnapshot(
|
||||
return result, err
|
||||
}
|
||||
|
||||
inLocalIndex, err := v.findSnapshotToRemove(snapshotID, opts)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
|
||||
if opts.DryRun {
|
||||
return v.removeSnapshotDryRun(result, snapshotID, opts)
|
||||
}
|
||||
@@ -1167,13 +1161,11 @@ func (v *Vaultik) RemoveSnapshot(
|
||||
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)
|
||||
if err != nil {
|
||||
return result, fmt.Errorf("removing from local database: %w", err)
|
||||
}
|
||||
err = v.deleteSnapshotFromLocalDB(snapshotID)
|
||||
if err != nil {
|
||||
return result, fmt.Errorf("removing from local database: %w", err)
|
||||
}
|
||||
|
||||
if !opts.LocalOnly {
|
||||
@@ -1192,11 +1184,9 @@ func (v *Vaultik) RemoveSnapshot(
|
||||
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 result.RemoteRemoved {
|
||||
if !opts.LocalOnly && result.RemoteRemoved {
|
||||
v.printlnStdout("Removed snapshot metadata from remote storage")
|
||||
v.stdoutf("\nNote: The removed snapshot's blobs remain on the remote. "+
|
||||
"Run '%s' to delete any blobs no longer referenced by any "+
|
||||
@@ -1206,45 +1196,6 @@ func (v *Vaultik) RemoveSnapshot(
|
||||
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
|
||||
// making changes.
|
||||
func (v *Vaultik) removeSnapshotDryRun(
|
||||
@@ -1304,14 +1255,14 @@ func (v *Vaultik) confirmRemoveSnapshot(snapshotID string, opts *RemoveOptions)
|
||||
// didn't finish and to run `vaultik snapshot remove` for the snapshot
|
||||
// again once the destination store is reachable (`vaultik prune` never
|
||||
// removes snapshot metadata). Returns true when the remote removal
|
||||
// succeeded and deleted something.
|
||||
// succeeded.
|
||||
func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) bool {
|
||||
log.Info("Removing snapshot metadata from remote storage",
|
||||
"snapshot_id", snapshotID)
|
||||
|
||||
remoteKey := snapshot.RemoteSnapshotKey(snapshotID)
|
||||
|
||||
deleted, err := v.deleteRemoteSnapshotByKey(remoteKey)
|
||||
err := v.deleteRemoteSnapshotByKey(remoteKey)
|
||||
if err != nil {
|
||||
log.Warn("Could not remove snapshot metadata from remote storage; "+
|
||||
"run '"+snapshotRemoveCommandHint+"' with the snapshot's ID "+
|
||||
@@ -1329,7 +1280,7 @@ func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) b
|
||||
return false
|
||||
}
|
||||
|
||||
return deleted > 0
|
||||
return true
|
||||
}
|
||||
|
||||
// RemoveAllSnapshots removes every snapshot known to the local
|
||||
@@ -1621,7 +1572,7 @@ func (v *Vaultik) removeAllLocalSnapshots(
|
||||
}
|
||||
|
||||
if !opts.LocalOnly {
|
||||
_, err = v.deleteRemoteSnapshotByKey(
|
||||
err = v.deleteRemoteSnapshotByKey(
|
||||
snapshot.RemoteSnapshotKey(snapshotID))
|
||||
if err != nil {
|
||||
log.Warn("Failed to remove snapshot metadata from remote",
|
||||
@@ -1647,7 +1598,7 @@ func (v *Vaultik) removeAllOrphanRemotes(
|
||||
for _, key := range orphanRemoteKeys {
|
||||
log.Info("Removing orphan remote snapshot", "remote_key", key)
|
||||
|
||||
_, err := v.deleteRemoteSnapshotByKey(key)
|
||||
err := v.deleteRemoteSnapshotByKey(key)
|
||||
if err != nil {
|
||||
log.Warn("Failed to remove orphan from remote",
|
||||
"remote_key", key, "error", err)
|
||||
@@ -1694,44 +1645,34 @@ func (v *Vaultik) deleteSnapshotFromLocalDB(snapshotID string) error {
|
||||
}
|
||||
|
||||
// deleteRemoteSnapshotByKey removes everything under
|
||||
// metadata/<remoteKey>/ on the destination store and returns how many
|
||||
// objects it deleted. The argument is a remote key (double-SHA256
|
||||
// derivation), not a human snapshot ID; callers that have a human ID
|
||||
// must hash via snapshot.RemoteSnapshotKey first.
|
||||
func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) (int, error) {
|
||||
objectsToDelete, err := v.remoteSnapshotObjects(remoteKey)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
// metadata/<remoteKey>/ on the destination store. The argument is a
|
||||
// remote key (double-SHA256 derivation), not a human snapshot ID;
|
||||
// callers that have a human ID must hash via snapshot.RemoteSnapshotKey
|
||||
// first.
|
||||
func (v *Vaultik) deleteRemoteSnapshotByKey(remoteKey string) error {
|
||||
prefix := fmt.Sprintf("metadata/%s/", 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 i, key := range objectsToDelete {
|
||||
err = v.Storage.Delete(v.ctx, key)
|
||||
for _, key := range objectsToDelete {
|
||||
err := v.Storage.Delete(v.ctx, key)
|
||||
if err != nil {
|
||||
return i, fmt.Errorf("removing %s: %w", key, err)
|
||||
return fmt.Errorf("removing %s: %w", key, err)
|
||||
}
|
||||
|
||||
log.Debug("Deleted remote object", "key", key)
|
||||
}
|
||||
|
||||
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
|
||||
return nil
|
||||
}
|
||||
|
||||
// outputRemoveJSON outputs the removal result as JSON
|
||||
|
||||
@@ -1,75 +0,0 @@
|
||||
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
|
||||
// bound to a different destination than the configured one.
|
||||
// bound to a different destination than the configured storage_url.
|
||||
var errStorageBindingMismatch = errors.New(
|
||||
"local index is bound to a different backup destination")
|
||||
|
||||
@@ -24,9 +24,8 @@ var errStorageBindingMismatch = errors.New(
|
||||
//
|
||||
// Behaviour:
|
||||
// - On first use (empty stored value), record the configured
|
||||
// destination (storage_url, or the s3.* endpoint, bucket and
|
||||
// prefix) and log the binding.
|
||||
// - When the stored value matches the configured destination, do
|
||||
// storage_url and log the binding.
|
||||
// - When the stored value matches the configured storage_url, do
|
||||
// nothing and return nil.
|
||||
// - When the two differ, refuse with an error that tells the user
|
||||
// how to recover (revert the config, or run `vaultik database
|
||||
@@ -47,10 +46,12 @@ func (v *Vaultik) EnsureStorageBinding() error {
|
||||
|
||||
configured := v.Config.StorageURL
|
||||
if configured == "" {
|
||||
// The s3.* form: bind on its endpoint, bucket and prefix,
|
||||
// written the way storage_url would name the same destination.
|
||||
s3 := v.Config.S3
|
||||
configured = "s3://" + s3.Bucket + "/" + s3.Prefix + "?endpoint=" + s3.Endpoint
|
||||
// Some legacy configs still use the split s3.* keys instead of
|
||||
// storage_url. Falling back to a synthetic URL for those would
|
||||
// hide the fact that the binding is loose. Instead, treat
|
||||
// unset as "nothing to bind against" — the check is
|
||||
// necessarily best-effort for those configs.
|
||||
return nil
|
||||
}
|
||||
|
||||
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
||||
@@ -95,8 +96,7 @@ func buildBindingMismatchMessage(stored, configured string) string {
|
||||
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
||||
"\n" +
|
||||
"To proceed, either:\n" +
|
||||
" - revert storage_url (or the s3.* endpoint, bucket and prefix) in your\n" +
|
||||
" config to the bound destination, or\n" +
|
||||
" - revert storage_url in your config to the bound destination, or\n" +
|
||||
" - run 'vaultik database delete' to discard the local index and rebuild it\n" +
|
||||
" from a fresh full backup against the new destination"
|
||||
}
|
||||
|
||||
@@ -76,51 +76,3 @@ func TestEnsureStorageBinding_MismatchRefuses(t *testing.T) {
|
||||
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
||||
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