Compare commits

..
Author SHA1 Message Date
sneak 2013d5655a Count a file as deleted only if the previous snapshot of its name had it (closes #304)
check / check (push) Waiting to run
A backup counted as deleted every file in the local index under the
source path that the walk did not find. That took in files the
snapshot excludes but another snapshot backs up, and files removed long
ago, which every later backup, and the first backup of another name
over the same path, reported as deleted again.

The backup now looks up the most recent completed snapshot of the same
name and counts only the files it lists that the walk did not find. A
file that is excluded, or inside an excluded directory, is not counted.

The snapshots row stores no deleted figure; only the backup's output
changes.

Model: opus-5-5
2026-10-13 14:23:18 +00:00
clawbot 005927a7fc Print snapshot list and purge sizes like every other command (closes #312)
check / check (push) Waiting to run
`snapshot list`, the `snapshot purge` listing and the `bytes` value in
terminal log lines divided by 1024 while labelling the result KB or MB,
so 6,329,198 bytes read 6.0 MB there and 6.3 MB in every other command.
All three now use humanize.Bytes, as the rest of the program does, and
the 1024-based helper in internal/vaultik is removed.

The size shared by the list tests is now 5,000,000 bytes, which the two
ways of counting print differently.

Judgement call: the log handler's copy of the formatter is changed too,
since the issue asks that every command print a byte count the same way.

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

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

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

Model: opus-5-5
2026-10-13 15:48:00 +02:00
17 changed files with 650 additions and 239 deletions
+4 -4
View File
@@ -279,8 +279,8 @@ Key methods:
- `Scan(ctx, path, snapshotID)` → Main scan operation
Scan phases:
1. **Phase 0**: Detect deleted files from previous snapshots
2. **Phase 1**: Walk directory, identify files needing processing
1. **Phase 1**: Walk directory, identify files needing processing
2. **Phase 1b**: Count as deleted the files that the previous snapshot of the same name lists and the walk did not find; a file that is excluded, or inside an excluded directory, is not counted
3. **Phase 2**: Process files (chunk → pack → upload)
#### SnapshotManager
@@ -320,13 +320,13 @@ CreateSnapshot(opts)
│ │
│ ├─► scanner.Scan(ctx, path, snapshotID)
│ │ │
│ │ ├─► Phase 0: detectDeletedFiles()
│ │ │
│ │ ├─► Phase 1: scanPhase()
│ │ │ Walk directory
│ │ │ Check file metadata changes
│ │ │ Build list of files to process
│ │ │
│ │ ├─► Phase 1b: detectDeletedFiles()
│ │ │
│ │ └─► Phase 2: processPhase()
│ │ For each file:
│ │ chunker.ChunkReaderStreaming()
+44
View File
@@ -277,6 +277,50 @@ func (r *FileRepository) ListUnderPath(
return files, rows.Err()
}
// ListUnderPathInSnapshot returns the files that ListUnderPath returns
// for path and that the snapshot lists, in no particular order.
func (r *FileRepository) ListUnderPathInSnapshot(
ctx context.Context, path, snapshotID string,
) ([]*File, error) {
path = strings.TrimRight(path, "/")
dirPrefix := path + "/"
query := `
SELECT f.id, f.path, f.source_path, f.mtime, f.mtime_nsec, f.size,
f.mode, f.uid, f.gid, f.link_target
FROM files f
JOIN snapshot_files sf ON sf.file_id = f.id
WHERE sf.snapshot_id = ?
AND (f.path = ? OR substr(f.path, 1, length(?)) = ?)
`
rows, err := r.db.conn.QueryContext(ctx, query,
snapshotID, path, dirPrefix, dirPrefix)
if err != nil {
return nil, fmt.Errorf("querying files: %w", err)
}
defer func() {
err := rows.Close()
if err != nil {
Fatalf("failed to close rows: %v", err)
}
}()
var files []*File
for rows.Next() {
file, err := r.scanFileRows(rows)
if err != nil {
return nil, fmt.Errorf("scanning file: %w", err)
}
files = append(files, file)
}
return files, rows.Err()
}
// ListIDsWithChunksNotInUploadedBlobs returns the IDs of the files whose
// path starts with prefix and that list at least one chunk held by no
// blob whose upload has completed (uploaded_ts set). A new snapshot
+59
View File
@@ -212,6 +212,65 @@ func TestFileRepositoryListUnderPath(t *testing.T) {
}
}
// ListUnderPathInSnapshot leaves out files outside path and files the
// snapshot does not list, though another snapshot lists them.
func TestFileRepositoryListUnderPathInSnapshot(t *testing.T) {
t.Parallel()
db, cleanup := setupTestDB(t)
defer cleanup()
ctx := context.Background()
files := database.NewFileRepository(db)
snapshots := database.NewSnapshotRepository(db)
listed := map[string][]string{
"one": {"/src/a.txt", "/src2/c.txt"},
"two": {"/src/b.txt"},
}
for snapshotID, paths := range listed {
err := snapshots.Create(ctx, nil, &database.Snapshot{
ID: types.SnapshotID(snapshotID),
Hostname: "host",
StartedAt: time.Now(),
})
if err != nil {
t.Fatalf("failed to create snapshot %s: %v", snapshotID, err)
}
for _, path := range paths {
err := files.Create(ctx, nil, &database.File{
Path: types.FilePath(path),
MTime: time.Now().Truncate(time.Second),
Mode: 0644,
})
if err != nil {
t.Fatalf("failed to create %s: %v", path, err)
}
err = snapshots.AddFile(ctx, nil, snapshotID, path)
if err != nil {
t.Fatalf("failed to add %s to %s: %v", path, snapshotID, err)
}
}
}
found, err := files.ListUnderPathInSnapshot(ctx, "/src", "one")
if err != nil {
t.Fatalf("failed to list files: %v", err)
}
got := make([]string, 0, len(found))
for _, f := range found {
got = append(got, f.Path.String())
}
if !slices.Equal(got, []string{"/src/a.txt"}) {
t.Errorf("got %q, want only /src/a.txt", got)
}
}
func TestFileRepositorySymlink(t *testing.T) {
t.Parallel()
+7 -10
View File
@@ -10,6 +10,8 @@ import (
"sync"
"time"
"unicode"
"github.com/dustin/go-humanize"
)
// groupSeparator joins an open group path to an attribute key. This
@@ -304,18 +306,13 @@ func formatDuration(d time.Duration) string {
}
}
// formatBytes formats bytes in a human-readable way
// formatBytes formats a byte count with humanize.Bytes, as every command
// prints one. humanize.Bytes takes no negative count, so one is printed
// as it is.
func formatBytes(b int64) string {
const unit = 1024
if b < unit {
if b < 0 {
return fmt.Sprintf("%d B", b)
}
div, exp := int64(unit), 0
for n := b / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.1f %cB", float64(b)/float64(div), "KMGTPE"[exp])
return humanize.Bytes(uint64(b))
}
+5 -4
View File
@@ -200,7 +200,8 @@ func TestTTYHandlerWithGroupQualifiesKeys(t *testing.T) {
func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
t.Parallel()
const oneAndAHalfKiB = 1536
// Printed as "1.0 MB", in powers of 1000 like every other byte count.
const oneMegabyte = 1_000_000
for name, testCase := range map[string]struct {
derive func(*slog.Logger) *slog.Logger
@@ -223,13 +224,13 @@ func TestTTYHandlerByteFormattingSurvivesGrouping(t *testing.T) {
var buf bytes.Buffer
logger := slog.New(log.NewTTYHandler(&buf, debugHandlerOptions()))
testCase.derive(logger).Info("uploaded", "bytes", oneAndAHalfKiB)
testCase.derive(logger).Info("uploaded", "bytes", oneMegabyte)
line := ansiEscape.ReplaceAllString(buf.String(), "")
assert.Contains(t, line, testCase.key+"=1.5 KB",
assert.Contains(t, line, testCase.key+"=1.0 MB",
"a byte count must be human-readable however it is qualified")
assert.NotContains(t, line, strconv.Itoa(oneAndAHalfKiB),
assert.NotContains(t, line, strconv.Itoa(oneMegabyte),
"the raw number must not survive the formatting")
})
}
+15 -11
View File
@@ -16,6 +16,9 @@ type ScannerParams struct {
Fs afero.Fs
Exclude []string // Exclude patterns (combined global + snapshot-specific)
SkipErrors bool // Skip file read errors (log loudly but continue)
// PreviousSnapshotID is the most recent completed snapshot of the
// same name, "" for none.
PreviousSnapshotID string
}
// Module exports backup functionality as an fx module.
@@ -44,17 +47,18 @@ func provideScannerFactory(
}
return NewScanner(ScannerConfig{
FS: params.Fs,
ChunkSize: cfg.ChunkSize.Int64(),
Repositories: repos,
Storage: storer,
MaxBlobSize: cfg.BlobSizeLimit.Int64(),
CompressionLevel: cfg.CompressionLevel,
AgeRecipients: cfg.AgeRecipients,
EnableProgress: params.EnableProgress,
UI: params.UI,
Exclude: excludes,
SkipErrors: params.SkipErrors,
FS: params.Fs,
ChunkSize: cfg.ChunkSize.Int64(),
Repositories: repos,
Storage: storer,
MaxBlobSize: cfg.BlobSizeLimit.Int64(),
CompressionLevel: cfg.CompressionLevel,
AgeRecipients: cfg.AgeRecipients,
EnableProgress: params.EnableProgress,
UI: params.UI,
Exclude: excludes,
SkipErrors: params.SkipErrors,
PreviousSnapshotID: params.PreviousSnapshotID,
})
}
}
+51 -19
View File
@@ -58,6 +58,10 @@ type Scanner struct {
compressionLevel int
ageRecipient string
snapshotID string // Current snapshot being processed
// previousSnapshotID is the most recent completed snapshot of the
// same name, "" for none. Only a file it lists can be counted as
// deleted.
previousSnapshotID string
// currentSourcePath is the source directory being scanned, stored with
// each file record.
currentSourcePath string
@@ -129,6 +133,9 @@ type ScannerConfig struct {
// SkipErrors skips files that cannot be opened or read (log loudly but
// continue); packer, database, encryption, and upload errors still abort.
SkipErrors bool
// PreviousSnapshotID is the most recent completed snapshot of the
// same name, "" for none.
PreviousSnapshotID string
}
// ScanResult contains the results of a scan operation. Files and bytes
@@ -204,6 +211,7 @@ func NewScanner(cfg ScannerConfig) *Scanner {
compiledExclude: compiledExclude,
progress: progress,
skipErrors: cfg.SkipErrors,
previousSnapshotID: cfg.PreviousSnapshotID,
ui: uiw,
pendingChunkHashes: make(map[string]struct{}),
}
@@ -255,8 +263,9 @@ func (s *Scanner) Scan(
filesToProcess := scanResult.FilesToProcess
// Phase 1b: Detect deleted files by comparing DB against scanned files
err = s.detectDeletedFilesFromMap(ctx, knownFiles, existingFiles, result)
// Phase 1b: Detect deleted files by comparing the previous snapshot
// of this name against scanned files
err = s.detectDeletedFiles(ctx, path, existingFiles, result)
if err != nil {
return nil, fmt.Errorf("detecting deleted files: %w", err)
}
@@ -436,9 +445,8 @@ func (s *Scanner) summarizeScanPhase(
}
// loadKnownFiles loads the known files at and beneath path from the
// database into a map for fast lookup. Every loaded file the scan does
// not find is counted as deleted. This avoids per-file database queries
// during the scan phase.
// database into a map for fast lookup. This avoids per-file database
// queries during the scan phase.
func (s *Scanner) loadKnownFiles(
ctx context.Context, path string,
) (map[string]*database.File, error) {
@@ -1938,21 +1946,28 @@ func (s *Scanner) queueFileForBatchInsert(
})
}
// detectDeletedFilesFromMap finds files that existed in previous snapshots
// but no longer exist. Uses pre-loaded maps to avoid any filesystem or
// database access.
func (s *Scanner) detectDeletedFilesFromMap(
// detectDeletedFiles counts as deleted each file at or beneath path that
// the previous snapshot of this name lists and the walk did not find.
// A file that is excluded, or inside an excluded directory, is not
// counted: the walk skips it whether or not it is still on disk.
func (s *Scanner) detectDeletedFiles(
ctx context.Context,
knownFiles map[string]*database.File,
path string,
existingFiles map[string]struct{},
result *ScanResult,
) error {
if len(knownFiles) == 0 {
if s.previousSnapshotID == "" {
return nil
}
// Check each known file against the enumerated set (no filesystem access needed)
for path, file := range knownFiles {
previousFiles, err := s.repos.Files.ListUnderPathInSnapshot(
ctx, path, s.previousSnapshotID)
if err != nil {
return fmt.Errorf("listing files of snapshot %s: %w",
s.previousSnapshotID, err)
}
for _, file := range previousFiles {
// Check context cancellation periodically
select {
case <-ctx.Done():
@@ -1960,13 +1975,16 @@ func (s *Scanner) detectDeletedFilesFromMap(
default:
}
// Check if the file exists in our enumerated set
if _, exists := existingFiles[path]; !exists {
// File has been deleted
result.FilesDeleted++
result.BytesDeleted += file.Size
log.Debug("Detected deleted file", "path", path, "size", file.Size)
filePath := file.Path.String()
_, exists := existingFiles[filePath]
if exists || s.excludedOrInExcludedDir(filePath, path) {
continue
}
result.FilesDeleted++
result.BytesDeleted += file.Size
log.Debug("Detected deleted file", "path", filePath, "size", file.Size)
}
if result.FilesDeleted > 0 {
@@ -1977,6 +1995,20 @@ func (s *Scanner) detectDeletedFilesFromMap(
return nil
}
// excludedOrInExcludedDir reports whether filePath, which is at or
// beneath rootPath, or any directory between rootPath and it, matches an
// exclude pattern. A pattern such as */.local/share/Trash matches the
// directory but not the files inside it, which the walk skips too.
func (s *Scanner) excludedOrInExcludedDir(filePath, rootPath string) bool {
for p := filePath; len(p) > len(rootPath); p = filepath.Dir(p) {
if s.shouldExclude(p, rootPath) {
return true
}
}
return false
}
// wrapPermissionError augments permission errors with platform-specific
// remediation instructions. On macOS, TCC-protected directories (Calendars,
// Reminders, Photos, etc.) return EPERM unless the running application has
+45 -1
View File
@@ -103,6 +103,8 @@ func (v *Vaultik) FetchAndDecryptBlob(
}
// FetchBlob downloads a blob and returns a reader for the encrypted data.
// The reader is closed as soon as ctx is cancelled, which ends a read
// blocked inside it.
func (v *Vaultik) FetchBlob(
ctx context.Context, blobHash string,
) (io.ReadCloser, error) {
@@ -120,5 +122,47 @@ func (v *Vaultik) FetchBlob(
return nil, fmt.Errorf("downloading blob %s: %w", shortHash(blobHash), err)
}
return rc, nil
return newCloseOnCancelReader(ctx, rc), nil
}
// closeOnCancelReader closes the reader it wraps when ctx is cancelled.
// Not every backend's reader stops on cancel by itself (the file
// backend's does not), and a read stalled on a slow or hung source would
// otherwise outlast the shutdown timeout, so the process would exit
// before restore or deep verify removed its decrypted temporary files.
type closeOnCancelReader struct {
io.ReadCloser
ctx context.Context //nolint:containedctx // Read reports its error
stop func() bool
}
func newCloseOnCancelReader(
ctx context.Context, rc io.ReadCloser,
) *closeOnCancelReader {
return &closeOnCancelReader{
ReadCloser: rc,
ctx: ctx,
stop: context.AfterFunc(ctx, func() { _ = rc.Close() }),
}
}
// Read returns the context's error, not the closed reader's, once ctx
// has been cancelled.
func (r *closeOnCancelReader) Read(p []byte) (int, error) {
n, err := r.ReadCloser.Read(p)
if err != nil && r.ctx.Err() != nil {
return n, r.ctx.Err()
}
return n, err
}
// Close closes the wrapped reader, unless the cancel already closed it.
func (r *closeOnCancelReader) Close() error {
if !r.stop() {
return nil
}
return r.ReadCloser.Close()
}
-115
View File
@@ -1,20 +1,15 @@
package vaultik_test
import (
"bytes"
"context"
"crypto/rand"
"io"
"os"
"path/filepath"
"slices"
"testing"
"github.com/spf13/afero"
"github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/config"
"sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/storage"
"sneak.berlin/go/vaultik/internal/ui"
"sneak.berlin/go/vaultik/internal/vaultik"
)
@@ -82,116 +77,6 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
"deep verify should fail on a corrupted blob")
}
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
// that deep verify reports under --json: on a healthy snapshot, with a
// damaged blob, and with missing blobs. Deep verify stops at the first
// damaged blob, so the blobs read before it are the ones verified; when
// blobs are missing it reads no blob contents, but counts every missing
// blob.
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
fs := afero.NewOsFs()
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "source")
storeDir := filepath.Join(tempDir, "remote")
dbPath := filepath.Join(tempDir, "index.sqlite")
chunkSize := int64(32 * 1024)
maxBlobSize := int64(128 * 1024)
// Random data does not compress or deduplicate, so it fills several
// blobs.
data := make([]byte, maxBlobSize*4)
_, err := rand.Read(data)
require.NoError(t, err)
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
require.NoError(t, afero.WriteFile(fs,
filepath.Join(dataDir, "data.bin"), data, 0o644))
ctx := context.Background()
cfg, storer, snapshotID := runFileStorageBackup(
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.NoError(t, err)
require.Equal(t, "ok", healthy.Status)
require.GreaterOrEqual(t, healthy.BlobCount, 3)
require.Equal(t, healthy.BlobCount, healthy.Verified)
require.Zero(t, healthy.Missing)
// Deep verify reads blobs in hash order, which is the sorted order of
// their storage keys.
keys, err := storer.List(ctx, "blobs/")
require.NoError(t, err)
slices.Sort(keys)
require.Len(t, keys, healthy.BlobCount)
// Damage the last blob read, keeping its length, so every blob before
// it is read and passes first.
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
blob, err := afero.ReadFile(fs, lastPath)
require.NoError(t, err)
blob[len(blob)/2] ^= 0xff
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Equal(t, healthy.BlobCount-1, got.Verified)
require.Zero(t, got.Missing)
// Delete the first blob and the damaged last one: both must be
// counted as missing, not only the first one found.
var deletedSize int64
for _, key := range []string{keys[0], keys[len(keys)-1]} {
info, statErr := storer.Stat(ctx, key)
require.NoError(t, statErr)
deletedSize += info.Size
require.NoError(t, storer.Delete(ctx, key))
}
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Equal(t, 2, got.Missing)
require.Equal(t, deletedSize, got.MissingSize)
require.Zero(t, got.Verified)
}
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
// the decoded result and the error RunDeepVerify returned.
func deepVerifyJSON(
ctx context.Context, t *testing.T,
cfg *config.Config, storer storage.Storer, snapshotID string,
) (vaultik.VerifyResult, error) {
t.Helper()
var out bytes.Buffer
v := &vaultik.Vaultik{
Config: cfg,
Storage: storer,
Fs: afero.NewOsFs(),
Stdout: &out,
Stderr: io.Discard,
UI: ui.NewWithColor(io.Discard, false),
}
v.SetContext(ctx)
err := v.RunDeepVerify(snapshotID,
&vaultik.VerifyOptions{Deep: true, JSON: true})
return decodeVerifyResult(t, out.Bytes()), err
}
// corruptOneBlob flips a middle byte of the first blob file found under
// blobsDir, leaving the file length unchanged.
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
+12 -11
View File
@@ -661,17 +661,18 @@ func faultScannerFactory(
) snapshot.ScannerFactory {
return func(params snapshot.ScannerParams) *snapshot.Scanner {
return snapshot.NewScanner(snapshot.ScannerConfig{
FS: params.Fs,
Storage: storer,
ChunkSize: faultChunkSize,
MaxBlobSize: faultMaxBlobSize,
CompressionLevel: cfg.CompressionLevel,
AgeRecipients: cfg.AgeRecipients,
Repositories: repos,
EnableProgress: params.EnableProgress,
UI: params.UI,
Exclude: params.Exclude,
SkipErrors: params.SkipErrors,
FS: params.Fs,
Storage: storer,
ChunkSize: faultChunkSize,
MaxBlobSize: faultMaxBlobSize,
CompressionLevel: cfg.CompressionLevel,
AgeRecipients: cfg.AgeRecipients,
Repositories: repos,
EnableProgress: params.EnableProgress,
UI: params.UI,
Exclude: params.Exclude,
SkipErrors: params.SkipErrors,
PreviousSnapshotID: params.PreviousSnapshotID,
})
}
}
-16
View File
@@ -89,22 +89,6 @@ type SnapshotInfo struct {
RemotePresent *bool `json:"remote_present"`
}
// formatBytes formats bytes in a human-readable format
func formatBytes(bytes int64) string {
const unit = 1024
if bytes < unit {
return fmt.Sprintf("%d B", bytes)
}
div, exp := int64(unit), 0
for n := bytes / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp])
}
// parseSnapshotTimestamp extracts the timestamp from a snapshot ID
// Format: hostname_snapshotname_2026-01-12T14:41:15Z
func parseSnapshotTimestamp(snapshotID string) (time.Time, error) {
+118 -2
View File
@@ -142,14 +142,14 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
}
// scratchEntries returns the vaultik blob-cache and snapshot-database
// scratch entries currently present in dir.
// scratch entries (restore's and deep verify's) currently present in dir.
func scratchEntries(t *testing.T, dir string) []string {
t.Helper()
var matches []string
for _, pattern := range []string{
"vaultik-blobcache-*", "vaultik-restore-*",
"vaultik-blobcache-*", "vaultik-restore-*", "vaultik-verify-*",
} {
found, err := filepath.Glob(filepath.Join(dir, pattern))
require.NoError(t, err)
@@ -206,3 +206,119 @@ func TestRestoreSkipErrorsCancelDuringBlobDownload(t *testing.T) {
assert.NotContains(t, out.String(), srcPath,
"the cancel was reported as a failed file")
}
// stallingBlobStorer wraps a Storer so that every blob download hands
// over the first half of the blob and then stalls, ignoring the context,
// as a file:// blob on a hung USB stick or network share does. Only
// closing the returned reader ends the stall.
type stallingBlobStorer struct {
storage.Storer
once sync.Once
stalled chan struct{}
}
func newStallingBlobStorer(inner storage.Storer) *stallingBlobStorer {
return &stallingBlobStorer{Storer: inner, stalled: make(chan struct{})}
}
func (s *stallingBlobStorer) Get(
ctx context.Context, key string,
) (io.ReadCloser, error) {
rc, err := s.Storer.Get(ctx, key)
if err != nil || !strings.HasPrefix(key, "blobs/") {
return rc, err
}
blob, err := io.ReadAll(rc)
_ = rc.Close()
if err != nil {
return nil, err
}
// The writer is never closed, so once the first half has been read
// the next Read blocks until the reader is closed.
pr, pw := io.Pipe()
go func() {
_, _ = pw.Write(blob[:len(blob)/2])
s.once.Do(func() { close(s.stalled) })
}()
return pr, nil
}
// TestInterruptDuringStalledBlobReadCleansTempDir stops a restore and a
// deep verify while a blob read is stalled on a source that ignores the
// context. Stopping must end that read, so the operation unwinds and
// removes its decrypted scratch files well within the shutdown timeout
// (issue #325).
//
// Not parallel: it points TMPDIR at a private directory (via t.Setenv)
// so it can assert on exactly the scratch files the operation created.
func TestInterruptDuringStalledBlobReadCleansTempDir(t *testing.T) {
log.Initialize(log.Config{})
operations := map[string]func(v *Vaultik, snapshotID, targetDir string) error{
"restore": func(v *Vaultik, snapshotID, targetDir string) error {
return v.Restore(&RestoreOptions{
SnapshotID: snapshotID,
TargetDir: targetDir,
})
},
"deep verify": func(v *Vaultik, snapshotID, _ string) error {
return v.RunDeepVerify(snapshotID, &VerifyOptions{Deep: true})
},
}
for name, op := range operations {
t.Run(name, func(t *testing.T) {
fs := afero.NewOsFs()
root := t.TempDir()
// Random content, so the blob stays large enough to stall
// part-way through its compressed, encrypted body.
cfg, storer, snapshotID, _ := backupOneFile(context.Background(),
t, fs, root, "a.bin", randomBytes(t, 256*1024), 0o644)
scratch := filepath.Join(root, "scratch")
require.NoError(t, fs.MkdirAll(scratch, 0o755))
t.Setenv("TMPDIR", scratch)
stalling := newStallingBlobStorer(storer)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
v := newRestoreVaultik(ctx, cfg, stalling, fs)
v.cancel = cancel
var opErr error
stop := v.StartOperation(func() {
opErr = op(v, snapshotID, filepath.Join(root, "restored"))
})
select {
case <-stalling.stalled:
case <-time.After(30 * time.Second):
t.Fatal("the operation never reached a blob read")
}
require.NotEmpty(t, scratchEntries(t, scratch),
"expected decrypted scratch files to exist mid-read")
stopCtx, stopCancel := context.WithTimeout(
context.Background(), 10*time.Second)
defer stopCancel()
require.True(t, stop(stopCtx),
"stop timed out; the stalled blob read was not interrupted")
require.ErrorIs(t, opErr, context.Canceled)
require.Empty(t, scratchEntries(t, scratch),
"decrypted scratch files remained after the interrupt")
})
}
}
+30 -6
View File
@@ -240,12 +240,18 @@ func (v *Vaultik) createNamedSnapshot(
return err
}
previousSnapshotID, err := v.previousSnapshotID(snapName)
if err != nil {
return err
}
scanner := v.ScannerFactory(snapshot.ScannerParams{
EnableProgress: !opts.Cron,
UI: v.UI,
Fs: v.Fs,
Exclude: v.Config.GetExcludes(snapName),
SkipErrors: opts.SkipErrors,
EnableProgress: !opts.Cron,
UI: v.UI,
Fs: v.Fs,
Exclude: v.Config.GetExcludes(snapName),
SkipErrors: opts.SkipErrors,
PreviousSnapshotID: previousSnapshotID,
})
snapshotID, err := v.SnapshotManager.CreateSnapshotWithName(
@@ -280,6 +286,24 @@ func (v *Vaultik) createNamedSnapshot(
return nil
}
// previousSnapshotID returns the ID of the most recent completed snapshot
// named snapName in the local index, or "" when there is none.
func (v *Vaultik) previousSnapshotID(snapName string) (string, error) {
snapshots, err := v.Repositories.Snapshots.ListRecent(v.ctx, listRecentLimit)
if err != nil {
return "", fmt.Errorf("listing snapshots: %w", err)
}
for _, s := range snapshots {
if s.CompletedAt != nil &&
parseSnapshotName(s.ID.String(), s.Hostname.String()) == snapName {
return s.ID.String(), nil
}
}
return "", nil
}
// resolveSnapshotPaths resolves source directories to absolute paths
// with symlink resolution.
func (v *Vaultik) resolveSnapshotPaths(snapName string) ([]string, error) {
@@ -633,7 +657,7 @@ func (v *Vaultik) confirmAndExecutePurge(
v.stdoutf(" %s (%s, %s)\n",
snap.ID,
snap.Timestamp.Format("2006-01-02 15:04:05"),
formatBytes(snap.CompressedSize))
ubytes(snap.CompressedSize))
}
}
@@ -0,0 +1,196 @@
package vaultik_test
import (
"bytes"
"context"
"fmt"
"os"
"path/filepath"
"testing"
"time"
"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/ui"
"sneak.berlin/go/vaultik/internal/vaultik"
)
// These tests cover https://git.eeqj.de/sneak/vaultik/issues/304: a
// backup counts a file as deleted only when the previous snapshot of
// the same name had it and it is gone, and never counts an excluded
// file.
// deletedCountEnv is a backup setup over the source directory src whose
// user-facing output is kept in out. Each test configures its snapshots
// with setSnapshot.
type deletedCountEnv struct {
v *vaultik.Vaultik
out *bytes.Buffer
src string
}
// newDeletedCountEnv writes files, keyed by path relative to src.
func newDeletedCountEnv(t *testing.T, files map[string][]byte) *deletedCountEnv {
t.Helper()
ctx := context.Background()
fs := afero.NewOsFs()
tempDir := t.TempDir()
srcDir := filepath.Join(tempDir, "src")
for rel, content := range files {
path := filepath.Join(srcDir, rel)
require.NoError(t, fs.MkdirAll(filepath.Dir(path), 0o755))
require.NoError(t, afero.WriteFile(fs, path, content, 0o644))
}
cfg := faultTestConfig()
cfg.IndexPath = filepath.Join(tempDir, "index.sqlite")
cfg.Snapshots = map[string]config.SnapshotConfig{}
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() })
repos := database.NewRepositories(db)
out := &bytes.Buffer{}
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
v.UI = ui.NewWithColor(out, false)
return &deletedCountEnv{v: v, out: out, src: srcDir}
}
// setSnapshot configures the snapshot name to back up the path rel
// beneath src, with the given excludes.
func (e *deletedCountEnv) setSnapshot(name, rel string, exclude ...string) {
e.v.Config.Snapshots[name] = config.SnapshotConfig{
Paths: []string{filepath.Join(e.src, rel)},
Exclude: exclude,
}
}
// backUp runs a backup of the named snapshot and returns its output.
func (e *deletedCountEnv) backUp(t *testing.T, name string) string {
t.Helper()
e.out.Reset()
require.NoError(t, e.v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
Snapshots: []string{name},
}))
return e.out.String()
}
// A file this snapshot excludes is not counted as deleted: neither one
// that another snapshot backs up, nor one that the previous snapshot of
// this name had before the exclude was added.
func TestBackupDoesNotCountExcludedFileAsDeleted(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
env := newDeletedCountEnv(t, map[string][]byte{
"top.txt": []byte("top"),
"cache/tmp.bin": []byte("cache"),
"photos/a.jpg": []byte("photo a"),
"photos/b.jpg": []byte("photo b"),
"photos/c.jpg": []byte("photo c"),
})
env.setSnapshot("photos", "photos")
env.setSnapshot("home", "", "/photos")
env.backUp(t, "photos")
assert.Contains(t, env.backUp(t, "home"), filesLine(2, 2, 0))
env.setSnapshot("home", "", "/photos", "/cache")
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
}
// A file inside a directory that an exclude added after a backup
// matches is not counted as deleted, though the pattern matches only the
// directory.
func TestBackupDoesNotCountFileInExcludedDirectoryAsDeleted(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
env := newDeletedCountEnv(t, map[string][]byte{
"top.txt": []byte("top"),
"user/.local/share/Trash/old.txt": []byte("old"),
})
env.setSnapshot("home", "")
assert.Contains(t, env.backUp(t, "home"), filesLine(2, 2, 0))
env.setSnapshot("home", "", "*/.local/share/Trash")
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
}
// A backup counts only files that the previous snapshot of its own name
// had, not one that a later snapshot of another name over the same path
// had and that is since removed.
func TestBackupIgnoresSnapshotOfAnotherName(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
env := newDeletedCountEnv(t, map[string][]byte{
"kept.txt": []byte("kept"),
})
env.setSnapshot("home", "")
env.setSnapshot("other", "")
env.backUp(t, "home")
// Snapshot start times are kept in whole seconds: wait so that other's
// snapshot is strictly the most recent.
time.Sleep(time.Second)
extra := filepath.Join(env.src, "extra.txt")
require.NoError(t, os.WriteFile(extra, []byte("extra"), 0o600))
env.backUp(t, "other")
require.NoError(t, os.Remove(extra))
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
}
// A file removed from disk is counted as deleted by the next backup of
// the snapshot that had it, and not by any later backup of that
// snapshot, nor by the first backup of another name over the same path.
func TestBackupCountsDeletedFileOnce(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
aContent := bytes.Repeat([]byte("a"), 1000)
bContent := bytes.Repeat([]byte("b"), 2000)
keptContent := bytes.Repeat([]byte("k"), 500)
env := newDeletedCountEnv(t, map[string][]byte{
"a.txt": aContent,
"b.txt": bContent,
"kept.txt": keptContent,
})
env.setSnapshot("home", "")
env.setSnapshot("other", "")
assert.Contains(t, env.backUp(t, "home"), filesLine(3, 3, 0))
require.NoError(t, os.Remove(filepath.Join(env.src, "a.txt")))
require.NoError(t, os.Remove(filepath.Join(env.src, "b.txt")))
summary := env.backUp(t, "home")
assert.Contains(t, summary,
"Files: 1 examined, 0 backed up, 1 unchanged, 2 deleted.")
assert.Contains(t, summary,
fmt.Sprintf("Data: %s total (%s backed up), %s deleted.",
env.v.UI.Size(int64(len(keptContent))), env.v.UI.Size(0),
env.v.UI.Size(int64(len(aContent)+len(bContent)))))
assert.Contains(t, env.backUp(t, "home"), filesLine(1, 0, 1))
assert.Contains(t, env.backUp(t, "other"), filesLine(1, 0, 1))
}
+3 -3
View File
@@ -553,8 +553,8 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
if snap.LocallyTracked {
id = snap.ID.String()
uncompressed = formatBytes(snap.UncompressedSize)
newChunks = formatBytes(snap.NewChunkSize)
uncompressed = ubytes(snap.UncompressedSize)
newChunks = ubytes(snap.NewChunkSize)
} else {
id = formatRemoteOnlyID(snap.RemoteKey)
uncompressed = remoteOnlyCell
@@ -564,7 +564,7 @@ func (v *Vaultik) printSnapshotTable(snapshots []SnapshotInfo) error {
_, err = fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
id,
snap.Timestamp.Format("2006-01-02 15:04:05"),
formatBytes(snap.CompressedSize),
ubytes(snap.CompressedSize),
uncompressed,
newChunks)
if err != nil {
+44 -2
View File
@@ -212,8 +212,8 @@ const (
listConfiguredName = "home"
listLocalID = "testhost_home_2026-03-01T10:00:00Z"
listRemoteID = "otherhost_media_2026-03-02T11:22:33Z"
// fiveMegabytes formats as "5.0 MB" through formatBytes.
fiveMegabytes = 5 * 1024 * 1024
// fiveMegabytes is printed as "5.0 MB" by every command.
fiveMegabytes = 5_000_000
)
// TestListSnapshots_RemoteWithoutSecretKey is the regression guard for
@@ -306,6 +306,48 @@ func TestListSnapshots_RemoteOnlyRowRendering(t *testing.T) {
"are not in the local index")
}
// TestSnapshotListAndPurgePrintSizesLikeRemoteInfo checks that the
// `snapshot list` table and the purge listing print a byte count the way
// `remote info` does: in powers of 1000, as the MB label says.
func TestSnapshotListAndPurgePrintSizesLikeRemoteInfo(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
const wantSize = "5.0 MB" // fiveMegabytes
env := newListEnv(t)
// A remote-only snapshot: `snapshot list` and `remote info` both take
// its size from the manifest.
remoteKey := env.addRemote(t, listRemoteID,
time.Date(2026, 3, 2, 11, 22, 33, 0, time.UTC))
// A locally tracked snapshot of the same size for the purge to list.
// The purge prints the snapshot's blob size, the last argument here.
start := time.Date(2026, 3, 1, 10, 0, 0, 0, time.UTC)
env.addLocal(t, listLocalID, start)
env.addRemote(t, listLocalID, start)
require.NoError(t, env.v.Repositories.Snapshots.UpdateCounts(
context.Background(), nil, listLocalID, 0, 0, 0, 0, fiveMegabytes))
require.NoError(t, env.v.RemoteInfo(false))
assert.Contains(t, findTableRow(t, env.stdout.String(), remoteKey[:16]),
wantSize, "remote info")
env.stdout.Reset()
require.NoError(t, env.v.ListSnapshots(false))
assert.Contains(t, findTableRow(t, env.stdout.String(),
"<remote only:"+remoteKey[:12]+">"), wantSize, "snapshot list")
// Without --force and with nothing on stdin the purge lists what it
// would delete, reads no confirmation, and deletes nothing.
env.stdout.Reset()
require.NoError(t, env.v.PurgeSnapshotsWithOptions(
&vaultik.SnapshotPurgeOptions{OlderThan: "1d"}))
assert.Contains(t, findTableRow(t, env.stdout.String(), listLocalID),
wantSize, "snapshot purge")
}
// findTableRow returns the single output line containing needle.
func findTableRow(t *testing.T, out, needle string) string {
t.Helper()
+17 -35
View File
@@ -32,9 +32,7 @@ var (
errManifestExtraBlob = errors.New("manifest contains blob not in database")
errManifestMissingBlob = errors.New(
"manifest omits blob present in database")
errBlobSizeMismatch = errors.New("blob size mismatch")
errBlobsNotInStorage = errors.New(
"blobs missing from storage or stored with the wrong size")
errBlobSizeMismatch = errors.New("blob size mismatch")
)
// verifyStatusFailed is the JSON status value for a failed verification.
@@ -137,6 +135,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
}
result.Status = "ok"
result.Verified = len(dbBlobs)
if opts.JSON {
return v.outputVerifyJSON(result)
@@ -291,7 +290,7 @@ func (v *Vaultik) runVerificationSteps(
v.stdoutf("Checking blob existence in remote storage...\n")
}
err = v.verifyBlobExistenceFromDB(dbBlobs, result)
err = v.verifyBlobExistenceFromDB(dbBlobs)
if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err)
}
@@ -302,8 +301,7 @@ func (v *Vaultik) runVerificationSteps(
len(dbBlobs), ubytes(totalSize))
}
err = v.performDeepVerificationFromDB(
dbBlobs, tdb.db.Conn(), opts, result, identities)
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err)
}
@@ -655,13 +653,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
return nil
}
// verifyBlobExistenceFromDB checks that every blob from the database is
// in storage with the size the database records. It checks them all,
// counting the missing and wrong-size blobs in result, and fails if it
// found any.
func (v *Vaultik) verifyBlobExistenceFromDB(
blobs []snapshot.BlobInfo, result *VerifyResult,
) error {
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
for i, blob := range blobs {
@@ -674,21 +667,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
// Construct blob path
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
// Check blob exists
stat, err := v.Storage.Stat(v.ctx, blobPath)
switch {
case err != nil:
log.Warn("Blob missing from storage",
"hash", blob.Hash, "error", err)
if err != nil {
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
}
result.Missing++
result.MissingSize += blob.CompressedSize
case stat.Size != blob.CompressedSize:
log.Warn("Blob has the wrong size in storage",
"hash", blob.Hash,
"storage_size", stat.Size,
"database_size", blob.CompressedSize)
result.Mismatched++
// Verify size matches
if stat.Size != blob.CompressedSize {
return fmt.Errorf(
"%w: blob %s: S3 has %d bytes, database has %d bytes",
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
}
// Progress update every 100 blobs
@@ -702,22 +691,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
}
}
if result.Missing > 0 || result.Mismatched > 0 {
return fmt.Errorf("%w: %d missing, %d with the wrong size",
errBlobsNotInStorage, result.Missing, result.Mismatched)
}
log.Info("✓ All blobs exist in storage")
return nil
}
// performDeepVerificationFromDB downloads and verifies the content of
// each blob using the database as source, stopping at the first blob that
// fails. Each blob that passes is counted in result.Verified.
// each blob using the database as source.
func (v *Vaultik) performDeepVerificationFromDB(
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
result *VerifyResult, identities []age.Identity,
identities []age.Identity,
) error {
// Calculate total bytes for ETA
var totalBytesExpected int64
@@ -740,7 +723,6 @@ func (v *Vaultik) performDeepVerificationFromDB(
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
}
result.Verified++
bytesProcessed += blobInfo.CompressedSize
elapsed := time.Since(startTime)
remaining := len(blobs) - (i + 1)