Author SHA1 Message Date
sneak 86ec0df2e7 Make an S3 listing the server cuts short an error (closes #297)
check / check (push) Waiting to run
The SDK paginator takes a listing page marked truncated but carrying no
continuation token for the last page and stops without an error. A
server answering V2 listings V1-style sends such pages, so ListObjects
and ListObjectsStream returned the first page as the whole listing, and
prune treated the blobs of the snapshots on later pages as unreferenced.
Both now fetch each page through nextListPage, which returns
ErrNoContinuationToken for such a page; prune, snapshot list and remote
nuke already handle a listing error.

The test serves a fixed reply from a stub server, since gofakes3 always
sends the token.

Model: opus-5-5
2026-10-13 00:04:07 +00:00
clawbot c9a519c233 Reject a retention duration longer than about 292 years (closes #289)
check / check (push) Waiting to run
parseDuration multiplied each number by the length of its unit with no
limit, so from 293y (or 15251w, 3559mo, 106752d, or a sum such as
200y200y) the result wrapped to a negative duration. The purge cutoff
then lay in the future and every snapshot of the name was selected, for
--older-than and --keep-newer-than alike. Each part is now checked
before it is added, and a total past the largest time.Duration is
rejected as an invalid duration, so nothing is deleted. Go units were
already safe: time.ParseDuration rejects an overflow.

With --prune the backup still runs before the duration is read; only
the prune step fails.

Model: opus-5-5
2026-10-13 01:51:24 +02:00
clawbot fa56165fe4 Stop a cancelled backup at once under --skip-errors (closes #286)
check / check (push) Canceled after 0s
After Ctrl-C or SIGTERM, phase 2 of a --skip-errors backup treated the
cancellation error from each remaining file like an unreadable file: it
opened the file, printed an error line, counted it as failed and moved
on to the next. The processing loop now checks for cancellation before
each file, and an error from a file once the run is cancelled stops the
run instead of being skipped. The loop check is what stops a run
cancelled while an empty file is open, since an empty file has no
chunks and nothing reads the context while it is backed up.

The --skip-errors test helper now takes the scan's context.

Model: opus-5-5
2026-10-08 15:12:07 +02:00
clawbot 34f61fb408 Make remote nuke delete leftover .partial uploads (closes #281)
check / check (push) Canceled after 0s
The file and rclone listings skip an object whose name ends in
`.partial`, the temporary name a `file://` or rclone upload writes
before moving the object into place. `remote nuke` deletes only what
the listings return, so it left the `.partial` objects killed uploads
leave behind and still reported the destination store empty.

Storer gains DeletePartialUploads. The file and rclone backends remove
every `.partial` object under the prefix; S3 has none to remove, since
it shows an object only once its upload completes. `remote nuke` calls
it for `metadata/` and `blobs/` as its last step.

Empty directories under a `file://` destination are still left behind.

Model: opus-5-5
2026-10-08 14:17:20 +02:00
clawbot 05bf73c48a Never skip a local index error for a directory or symlink (closes #284)
check / check (push) Canceled after 0s
Under --skip-errors, an error recording a directory or symlink in the
local index was skipped like an unreadable file. The snapshot completed
without the entry, and a restore did not recreate it.

processFileWithErrorHandling now records a directory or symlink itself
and returns any error, which stops the backup with or without the flag.
The skip sees only a regular file's errors, so countFailedFile no longer
checks for a directory.

The test injects the error with a SQLite trigger that refuses the
directory's files row.

Model: opus-5-5
2026-10-08 12:59:26 +02:00
clawbot 79a73fa122 Count a file a backup could not store as failed (closes #280)
check / check (push) Canceled after 0s
A file that phase 1 of a backup counted and phase 2 could not open,
because it was unreadable under --skip-errors or removed in between,
was added to the unchanged count while its size stayed in
BytesScanned. The summary showed it as unchanged with its bytes backed
up, and the snapshots row's file_count and total_size included it.

The scanner now counts such a file in FilesFailed and takes its size
out of BytesScanned. The summary's files line adds "N failed", and
file_count leaves the file out.

A directory phase 2 cannot record is not counted as failed, since
phase 1 counts no directories; that case has no test.

Model: opus-5-5
2026-10-08 10:12:10 +02:00
21 changed files with 647 additions and 40 deletions
+43
View File
@@ -22,6 +22,49 @@ 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-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
([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
+29 -2
View File
@@ -18,6 +18,13 @@ 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
@@ -214,7 +221,7 @@ func (c *Client) ListObjects(ctx context.Context, prefix string) ([]string, erro
})
for paginator.HasMorePages() {
page, err := paginator.NextPage(ctx)
page, err := nextListPage(ctx, paginator)
if err != nil {
return nil, err
}
@@ -235,6 +242,26 @@ 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.
@@ -300,7 +327,7 @@ func (c *Client) ListObjectsStream(
})
for paginator.HasMorePages() {
page, err := paginator.NextPage(ctx)
page, err := nextListPage(ctx, paginator)
if err != nil {
ch <- ObjectInfo{Err: err}
+30 -14
View File
@@ -1312,6 +1312,13 @@ 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)
@@ -1348,13 +1355,32 @@ func (s *Scanner) processPhase(
return s.finalizeProcessPhase(ctx, result)
}
// processFileWithErrorHandling wraps processFileStreaming with error recovery for
// deleted files and skip-errors mode. Returns (skipped, error).
// 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).
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
@@ -1392,12 +1418,7 @@ 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()
}
@@ -1770,16 +1791,11 @@ func (e *packerError) Error() string { return e.err.Error() }
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(
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))
+90 -11
View File
@@ -82,6 +82,32 @@ 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.
@@ -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,
// printing user-facing messages to uiw (nil discards them), and returns the
// repositories (for inspection) and the scan error.
// 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.
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) {
t.Helper()
db, err := database.NewTestDB()
db, err := database.New(ctx, ":memory:")
if err != nil {
t.Fatalf("create test db: %v", err)
}
@@ -161,7 +188,6 @@ func runSkipErrorScan(
SkipErrors: skipErrors,
})
ctx := context.Background()
snapshotID := "test-snapshot-skip-errors"
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/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 {
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}
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 {
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}
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 {
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)
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) {
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}
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 {
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)
}
}
// 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())
}
})
}
}
@@ -171,6 +171,11 @@ func (f *Storer) ListStream(
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.
func (f *Storer) Info() storage.Info {
return f.inner.Info()
+36
View File
@@ -242,6 +242,42 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
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.
func (f *FileStorer) Info() Info {
return Info{
+42
View File
@@ -120,3 +120,45 @@ func TestFileStorer_ListSkipsPartialFiles(t *testing.T) {
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)
}
}
+25
View File
@@ -213,6 +213,31 @@ func (r *RcloneStorer) ListStream(
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.
func (r *RcloneStorer) Info() Info {
location := r.remote
+41
View File
@@ -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
// 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
+6
View File
@@ -99,6 +99,12 @@ func (s *S3Storer) ListStream(ctx context.Context, prefix string) <-chan ObjectI
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.
func (s *S3Storer) Info() Info {
return Info{
+69 -3
View File
@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"slices"
@@ -22,13 +23,16 @@ import (
// s3TestBucket is the bucket created for each in-process S3 server.
const s3TestBucket = "test-bucket"
// Credentials for the tests that build a storer from a config.Config. The
// in-process S3 server accepts any.
// Credentials for the tests' S3 clients. The test servers accept 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,
@@ -55,7 +59,7 @@ func newS3Storer(t *testing.T) storage.Storer {
Bucket: s3TestBucket,
AccessKeyID: "test",
SecretAccessKey: "test",
Region: "us-east-1",
Region: s3TestRegion,
})
if err != nil {
t.Fatalf("new client: %v", err)
@@ -93,6 +97,68 @@ 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
+6
View File
@@ -71,6 +71,12 @@ type Storer interface {
// If an error occurs during listing, the final item will have Err set.
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() Info
}
@@ -257,6 +257,10 @@ func (s *stubLister) List(_ context.Context, _ string) ([]string, error) {
return nil, errStubUnused
}
func (s *stubLister) DeletePartialUploads(_ context.Context, _ string) error {
return errStubUnused
}
func (s *stubLister) Info() storage.Info {
return storage.Info{}
}
+24 -8
View File
@@ -3,6 +3,7 @@ package vaultik
import (
"errors"
"fmt"
"math"
"regexp"
"strconv"
"strings"
@@ -47,6 +48,10 @@ 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
@@ -145,10 +150,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, 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).
// 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).
func parseDuration(s string) (time.Duration, error) {
if strings.HasPrefix(strings.TrimSpace(s), "-") {
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)
}
var unitLength time.Duration
unit := strings.ToLower(match[2])
switch unit {
case "d", "day", "days":
total += time.Duration(n) * day
unitLength = day
case "w", "week", "weeks":
total += time.Duration(n) * week
unitLength = week
case "mo", "month", "months":
total += time.Duration(n) * month
unitLength = month
case "y", "year", "years":
total += time.Duration(n) * year
unitLength = 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
+11
View File
@@ -111,6 +111,17 @@ 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 {
+6
View File
@@ -155,6 +155,12 @@ func (m *MockStorer) ListStream(
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 {
return storage.Info{
Type: "mock",
+64
View File
@@ -0,0 +1,64 @@
package vaultik_test
import (
"context"
"io/fs"
"os"
"path/filepath"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/snapshot"
)
// 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, _ := 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)
}
+12 -2
View File
@@ -24,8 +24,9 @@ var errNukeRequiresForce = errors.New(
const metadataDirName = "metadata"
// NukeRemote deletes every snapshot's metadata and every blob from remote
// storage. After this returns successfully the bucket prefix is empty and
// the next backup starts from scratch.
// storage, along with any object an upload cut off part-way left under a
// 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
// confirming with the user.
@@ -48,6 +49,15 @@ func (v *Vaultik) NukeRemote(force bool) error {
return fmt.Errorf("pruning blobs: %w", err)
}
// 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)
}
}
v.UI.Completef("Backup destination store is now empty.")
return nil
+23
View File
@@ -267,6 +267,29 @@ 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()
+6
View File
@@ -125,6 +125,12 @@ func (s *testStorer) ListStream(
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 {
return storage.Info{
Type: testLabel,
@@ -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)
}