1 Commits
Author SHA1 Message Date
sneak 52def7b6ab Refuse credentials in a storage URL's endpoint or rclone remote name (closes #322)
check / check (push) Canceled after 0s
storage_url refused only the outer URL's userinfo and unknown query
parameters. Userinfo inside the s3 endpoint= value, and rclone's
remote,param=value parameters in the remote name, were accepted and
then printed by info, remote info and snapshot create. Both are now
refused without echoing the value: a host name never contains "@", and
an rclone remote name never contains a comma. An s3 or rclone query
that cannot be decoded (a ";" or a bad "%" escape) is refused too,
because decoding drops that part before it can be checked. The comma
is checked before url.Parse, whose error for such a name prints part of
it. The rclone remote name case has its own error pointing at rclone
config.

Model: opus-5-5
2026-10-13 23:47:02 +00:00
13 changed files with 156 additions and 379 deletions
+9 -13
View File
@@ -360,20 +360,16 @@ snapshot from the local index and strips the snapshot's metadata from
the backup destination store. Blobs are NOT touched — deleting blobs the backup destination store. Blobs are NOT touched — deleting blobs
requires reading every remaining remote manifest (the destination store requires reading every remaining remote manifest (the destination store
may hold snapshots this host doesn't know about), which is what may hold snapshots this host doesn't know about), which is what
`vaultik prune` does. The command prints the `vaultik prune` invocation `vaultik prune` does. On success the command prints the exact `vaultik
to run as a follow-up only when it removed the snapshot's metadata from prune` invocation to run as a follow-up. An ID that names no snapshot in
the destination store, and never under `--json`. An ID that names no the local index or on the destination store is an error; with
snapshot in the local index or on the destination store is an error; `--local-only`, the snapshot must be in the local index. Local row
with `--local-only`, the snapshot must be in the local index. Local row
cleanup (files, chunks, blobs the snapshot was the last referrer for) cleanup (files, chunks, blobs the snapshot was the last referrer for)
runs automatically. If removing the snapshot's metadata from the runs automatically. If the destination store is unreachable, the local-DB
destination store fails (for example because the store is unreachable), removal still completes and a warning is emitted; run `vaultik snapshot
the local-DB removal still completes and a warning is emitted; run remove <snapshot-id>` again once the store is reachable to remove the
`vaultik snapshot remove <snapshot-id>` again once the store is reachable snapshot's metadata from it (`vaultik prune` does not). To wipe everything
to remove the snapshot's metadata from it (`vaultik prune` does not). For on the destination in one go, use `vaultik remote nuke --force`.
a snapshot that is only on the destination store, that failure is an
error. To wipe everything on the destination in one go, use `vaultik
remote nuke --force`.
* `--local-only`: Skip remote cleanup; only touch the local index * `--local-only`: Skip remote cleanup; only touch the local index
* `--dry-run`: Show what would be deleted without deleting * `--dry-run`: Show what would be deleted without deleting
* `--force`: Skip confirmation prompt * `--force`: Skip confirmation prompt
+6 -9
View File
@@ -270,9 +270,8 @@ By default, this removes the snapshot from the local index database and
strips the snapshot's metadata from the backup destination store. Blobs strips the snapshot's metadata from the backup destination store. Blobs
are NOT touched: deleting them requires reading every remaining remote are NOT touched: deleting them requires reading every remaining remote
manifest (the destination store may hold snapshots this host doesn't manifest (the destination store may hold snapshots this host doesn't
know about), which is what 'vaultik prune' does. The command prints the know about), which is what 'vaultik prune' does. On success the command
'vaultik prune' invocation to run as a follow-up only when it removed the prints the exact 'vaultik prune' invocation to run as a follow-up.
snapshot's metadata from the destination store, and never under --json.
An ID that names no snapshot in the local index or on the destination An ID that names no snapshot in the local index or on the destination
store is an error. store is an error.
@@ -281,12 +280,10 @@ 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 snapshot locally without touching the destination store); the snapshot
must then be in the local index. must then be in the local index.
If removing the snapshot's metadata from the destination store fails (for If the remote is unreachable, the local-database removal still completes
example because the remote is unreachable), the local-database removal and a warning is emitted; run 'vaultik snapshot remove <snapshot-id>' again
still completes and a warning is emitted; run 'vaultik snapshot remove once the destination store is reachable to remove the snapshot's metadata
<snapshot-id>' again once the destination store is reachable to remove from it ('vaultik prune' does not).
the snapshot's metadata from it ('vaultik prune' does not). For a
snapshot that is only on the destination store, that failure is an error.
To wipe the entire destination store and start over, use 'vaultik remote To wipe the entire destination store and start over, use 'vaultik remote
nuke --force' — it is the single supported entry point for that.`, nuke --force' — it is the single supported entry point for that.`,
+3 -3
View File
@@ -39,9 +39,9 @@ type blobDiskCacheEntry struct {
// internally and will fall back to LRU eviction if curBytes exceeds // internally and will fall back to LRU eviction if curBytes exceeds
// maxBytes. Restore passes math.MaxInt64 as maxBytes and drives // maxBytes. Restore passes math.MaxInt64 as maxBytes and drives
// eviction itself via Delete() through restoreSweeper, which deletes // eviction itself via Delete() through restoreSweeper, which deletes
// each blob once no file still to be restored in this run needs it. // each blob the moment every file that references its chunks has been
// LRU never fires under that configuration; it is kept as a safety net // written. LRU never fires under that configuration; it is kept as a
// for callers that don't manage eviction themselves. // safety net for callers that don't manage eviction themselves.
// //
// Get/ReadAt/peak-Len counters are debugging instrumentation used by // Get/ReadAt/peak-Len counters are debugging instrumentation used by
// tests to assert that the restore code path uses ReadAt rather than // tests to assert that the restore code path uses ReadAt rather than
-60
View File
@@ -1,17 +1,13 @@
package vaultik //nolint:testpackage // exercises unexported count helpers package vaultik //nolint:testpackage // exercises unexported count helpers
import ( import (
"bytes"
"context" "context"
"testing" "testing"
"time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/database" "sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/log" "sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/snapshot"
"sneak.berlin/go/vaultik/internal/ui"
) )
// TestTableCountForReportSurfacesReadFailure is the regression guard for // TestTableCountForReportSurfacesReadFailure is the regression guard for
@@ -80,59 +76,3 @@ func TestCountDiffUnknownWhenEitherSideUnknown(t *testing.T) {
assert.Nil(t, countDiff(&before, nil), "unknown after yields unknown delta") assert.Nil(t, countDiff(&before, nil), "unknown after yields unknown delta")
assert.Nil(t, countDiff(nil, nil)) assert.Nil(t, countDiff(nil, nil))
} }
// TestPruneDatabaseSummaryLabelsTotals removes the only incomplete
// snapshot from an index that also holds a complete one. The figure in
// brackets is every snapshot left in the local index, so the line must
// say so rather than read as incomplete snapshots that remain.
func TestPruneDatabaseSummaryLabelsTotals(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
ctx := context.Background()
db, err := database.New(ctx, ":memory:")
require.NoError(t, err)
t.Cleanup(func() { _ = db.Close() })
repos := database.NewRepositories(db)
startedAt := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
completedAt := startedAt.Add(time.Minute)
require.NoError(t, repos.Snapshots.Create(ctx, nil, &database.Snapshot{
ID: "host_complete_2026-01-02T03:04:05Z",
Hostname: "host",
StartedAt: startedAt,
CompletedAt: &completedAt,
}))
require.NoError(t, repos.Snapshots.Create(ctx, nil, &database.Snapshot{
ID: "host_incomplete_2026-01-02T03:04:05Z",
Hostname: "host",
StartedAt: startedAt,
}))
var out bytes.Buffer
v := &Vaultik{
DB: db,
Repositories: repos,
SnapshotManager: snapshot.NewSnapshotManager(
snapshot.SnapshotManagerParams{Repos: repos}),
UI: ui.NewWithColor(&out, false),
}
v.SetContext(ctx)
result, err := v.PruneDatabase()
require.NoError(t, err)
require.Equal(t, int64(1), result.SnapshotsDeleted)
assert.Contains(t, out.String(), "Incomplete snapshots: 1 removed "+
"(total snapshots left in the local index: 1).")
assert.Contains(t, out.String(), "Orphaned files: 0 removed "+
"(total files left in the local index: 0).")
assert.Contains(t, out.String(), "Orphaned chunks: 0 removed "+
"(total chunks left in the local index: 0).")
assert.Contains(t, out.String(), "Orphaned blobs: 0 removed "+
"(total blobs left in the local index: 0).")
}
-35
View File
@@ -17,7 +17,6 @@ import (
"sneak.berlin/go/vaultik/internal/log" "sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/snapshot" "sneak.berlin/go/vaultik/internal/snapshot"
"sneak.berlin/go/vaultik/internal/storage" "sneak.berlin/go/vaultik/internal/storage"
"sneak.berlin/go/vaultik/internal/storage/faultstore"
"sneak.berlin/go/vaultik/internal/types" "sneak.berlin/go/vaultik/internal/types"
"sneak.berlin/go/vaultik/internal/vaultik" "sneak.berlin/go/vaultik/internal/vaultik"
) )
@@ -432,40 +431,6 @@ func TestRemoveSnapshot_NotInLocalIndex(t *testing.T) {
assert.Contains(t, out, "Removed snapshot metadata from remote storage") assert.Contains(t, out, "Removed snapshot metadata from remote storage")
} }
// TestRemoveSnapshot_NotInLocalIndexDeleteFails checks that removing a
// snapshot only the destination store holds fails when its metadata
// cannot be deleted there, under --force and under --json: nothing was
// removed, so the command must not succeed.
func TestRemoveSnapshot_NotInLocalIndexDeleteFails(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
for name, opts := range map[string]vaultik.RemoveOptions{
"force": {Force: true},
"json": {JSON: true},
} {
t.Run(name, func(t *testing.T) {
t.Parallel()
store := newTestStorer()
addManifest(t, store, "snapshot-001", nil)
faulty := faultstore.New(store)
faulty.OnDelete = func(string) bool { return true }
tv := vaultik.NewForTesting(faulty)
tv.Repositories = newLocalIndex(t)
result, err := tv.RemoveSnapshot("snapshot-001", &opts)
require.ErrorIs(t, err, faultstore.ErrInjectedDelete)
assert.False(t, result.RemoteRemoved)
assert.True(t, store.hasKey(remoteKeyPath("snapshot-001")))
assert.Empty(t, tv.Stdout.String())
})
}
}
func TestRemoveAllSnapshots_RequiresForce(t *testing.T) { func TestRemoveAllSnapshots_RequiresForce(t *testing.T) {
log.Initialize(log.Config{}) log.Initialize(log.Config{})
t.Parallel() t.Parallel()
+10 -8
View File
@@ -323,6 +323,12 @@ func (v *Vaultik) restoreAllFiles(
_ = blobCache.Close() _ = blobCache.Close()
}() }()
// Per-restore sweep state: every blob_size_limit/100 bytes written,
// scan the cache and delete any blob whose remaining file references
// are all already restored.
sweeper := newRestoreSweeper(v.ctx, repos, blobCache,
v.Config.BlobSizeLimit.Int64()/sweepIntervalDivisor)
blobByHash, blobIDToHash, err := v.buildBlobIndexes(repos) blobByHash, blobIDToHash, err := v.buildBlobIndexes(repos)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -334,12 +340,6 @@ func (v *Vaultik) restoreAllFiles(
return nil, fmt.Errorf("building restore plan: %w", err) return nil, fmt.Errorf("building restore plan: %w", err)
} }
// Per-restore sweep state: every blob_size_limit/100 bytes written,
// scan the cache and delete any blob that no pending file in the
// plan references.
sweeper := newRestoreSweeper(plan, blobCache,
v.Config.BlobSizeLimit.Int64()/sweepIntervalDivisor)
// Ranging over files rather than the map reports them in a stable order. // Ranging over files rather than the map reports them in a stable order.
for _, file := range files { for _, file := range files {
if planErr, ok := unplanned[file.ID]; ok { if planErr, ok := unplanned[file.ID]; ok {
@@ -427,8 +427,10 @@ func (v *Vaultik) runRestoreLoop(
continue continue
} }
// Drop the file from the plan's indexes so future picks ignore // Record the file as restored so the sweeper can free blobs
// it and the sweeper can free blobs no pending file needs. // once all referencing files are done, and drop it from the
// plan's indexes so future picks ignore it.
session.sweeper.fileRestored(fileID.String())
plan.finishFile(fileID) plan.finishFile(fileID)
processed++ processed++
+1 -2
View File
@@ -74,8 +74,7 @@ func TestRestoreCleansTempDirOnInterrupt(t *testing.T) {
buildLocalityFixture(t, fs, dataDir) buildLocalityFixture(t, fs, dataDir)
cfg, storer, snapshotID := setupLocalityBackup( cfg, storer, snapshotID := setupLocalityBackup(
context.Background(), t, fs, dataDir, storeDir, dbPath, context.Background(), t, fs, dataDir, storeDir, dbPath)
localitySourceBytes)
// Point the "" temp paths (the blob cache directory and the // Point the "" temp paths (the blob cache directory and the
// snapshot-database directory) at a private directory so the test can // snapshot-database directory) at a private directory so the test can
+11 -15
View File
@@ -63,10 +63,6 @@ type localityCopy struct {
data []byte data []byte
} }
// localitySourceBytes is the size of each source file of the locality
// fixture.
const localitySourceBytes = 1024 * 1024
// buildLocalityFixture writes the adversarial source layout described // buildLocalityFixture writes the adversarial source layout described
// in TestRestoreLocalityAndReadAt: 15 sources of 1 MiB each (which the // in TestRestoreLocalityAndReadAt: 15 sources of 1 MiB each (which the
// backup packs into 3 blobs of 5 chunks) plus 9 interleaved-name copies // backup packs into 3 blobs of 5 chunks) plus 9 interleaved-name copies
@@ -77,6 +73,7 @@ func buildLocalityFixture(
t.Helper() t.Helper()
const ( const (
srcBytes = 1024 * 1024
srcCount = 15 srcCount = 15
blobsCount = 3 blobsCount = 3
perBlob = srcCount / blobsCount perBlob = srcCount / blobsCount
@@ -86,7 +83,7 @@ func buildLocalityFixture(
for i := range srcCount { for i := range srcCount {
s := &localitySource{ s := &localitySource{
path: fmt.Sprintf("src-%03d.bin", i+1), path: fmt.Sprintf("src-%03d.bin", i+1),
data: randomBytes(t, localitySourceBytes), data: randomBytes(t, srcBytes),
} }
sources[i] = s sources[i] = s
require.NoError(t, afero.WriteFile( require.NoError(t, afero.WriteFile(
@@ -118,20 +115,19 @@ func buildLocalityFixture(
// setupLocalityBackup runs the backup half of the locality test: a // setupLocalityBackup runs the backup half of the locality test: a
// snapshot of dataDir into a file storer with sizes tuned so 15 one-chunk // snapshot of dataDir into a file storer with sizes tuned so 15 one-chunk
// files of srcBytes pack into 3 blobs. The index database is closed // files pack into 3 blobs. The index database is closed before returning
// before returning so the restore half runs from remote bytes only. // so the restore half runs from remote bytes only.
func setupLocalityBackup( func setupLocalityBackup(
ctx context.Context, t *testing.T, fs afero.Fs, ctx context.Context, t *testing.T, fs afero.Fs,
dataDir, storeDir, dbPath string, srcBytes int64, dataDir, storeDir, dbPath string,
) (*config.Config, *storage.FileStorer, string) { ) (*config.Config, *storage.FileStorer, string) {
t.Helper() t.Helper()
// An average chunk size of four times srcBytes makes the minimum // chunkSize avg = 4 MiB makes minSize = 1 MiB, so a 1 MiB file
// chunk srcBytes, so each source file becomes one chunk. A blob limit // becomes one chunk. maxBlobSize = 5 MiB packs exactly 5 chunks
// of five times srcBytes packs exactly 5 chunks per blob, yielding 3 // per blob, yielding 3 blobs from 15 source files.
// blobs from 15 source files. chunkSize := int64(4 * 1024 * 1024)
chunkSize := 4 * srcBytes maxBlobSize := int64(5 * 1024 * 1024)
maxBlobSize := 5 * srcBytes
storer, err := storage.NewFileStorer(storeDir) storer, err := storage.NewFileStorer(storeDir)
require.NoError(t, err) require.NoError(t, err)
@@ -212,7 +208,7 @@ func TestRestoreLocalityAndReadAt(t *testing.T) {
ctx := context.Background() ctx := context.Background()
cfg, storer, snapshotID := setupLocalityBackup( cfg, storer, snapshotID := setupLocalityBackup(
ctx, t, fs, dataDir, storeDir, dbPath, localitySourceBytes) ctx, t, fs, dataDir, storeDir, dbPath)
// Wrap the storer so we can count downloads per blob key. // Wrap the storer so we can count downloads per blob key.
counter := newCountingStorer(storer) counter := newCountingStorer(storer)
+1 -8
View File
@@ -31,8 +31,7 @@ var (
// with no further downloads. // with no further downloads.
// - blobFiles: for each blob, the set of pending files referencing // - blobFiles: for each blob, the set of pending files referencing
// it. Used to short-circuit "when this blob lands, which files // it. Used to short-circuit "when this blob lands, which files
// become ready" without a global scan, and by the sweeper to tell // become ready" without a global scan.
// whether a cached blob is still needed.
type restorePlan struct { type restorePlan struct {
fileBlobs map[types.FileID]map[string]struct{} fileBlobs map[types.FileID]map[string]struct{}
blobFiles map[string]map[types.FileID]struct{} blobFiles map[string]map[types.FileID]struct{}
@@ -238,12 +237,6 @@ func (p *restorePlan) filesReferencingBlob(blobHash string) []types.FileID {
return out return out
} }
// pendingFileUsesBlob reports whether any pending file references the
// named blob, whether or not the blob is cached.
func (p *restorePlan) pendingFileUsesBlob(blobHash string) bool {
return len(p.blobFiles[blobHash]) > 0
}
// hasPending reports whether any unfinished files remain. // hasPending reports whether any unfinished files remain.
func (p *restorePlan) hasPending() bool { func (p *restorePlan) hasPending() bool {
return len(p.fileBlobs) > 0 return len(p.fileBlobs) > 0
+9 -9
View File
@@ -63,6 +63,15 @@ func TestRunRestoreLoopFailsOnAbandonedFiles(t *testing.T) {
t.Cleanup(func() { _ = cache.Close() }) t.Cleanup(func() { _ = cache.Close() })
v := &Vaultik{ctx: ctx}
session := &restoreSession{
v: v,
ctx: ctx,
repos: repos,
sweeper: newRestoreSweeper(ctx, repos, cache, 1),
result: &RestoreResult{},
}
// A file that is still pending but whose uncached-blob set is empty // A file that is still pending but whose uncached-blob set is empty
// and which was never queued as ready: the loop can neither restore // and which was never queued as ready: the loop can neither restore
// nor download it. This is the abandonment the guard must catch. // nor download it. This is the abandonment the guard must catch.
@@ -74,15 +83,6 @@ func TestRunRestoreLoopFailsOnAbandonedFiles(t *testing.T) {
cached: map[string]struct{}{}, cached: map[string]struct{}{},
} }
v := &Vaultik{ctx: ctx}
session := &restoreSession{
v: v,
ctx: ctx,
repos: repos,
sweeper: newRestoreSweeper(plan, cache, 1),
result: &RestoreResult{},
}
err = v.runRestoreLoop(session, plan, map[types.FileID]*database.File{}, 0) err = v.runRestoreLoop(session, plan, map[types.FileID]*database.File{}, 0)
require.ErrorIs(t, err, errRestoreIncomplete) require.ErrorIs(t, err, errRestoreIncomplete)
} }
+88 -14
View File
@@ -1,30 +1,46 @@
package vaultik package vaultik
// restoreSweeper frees cached blobs once no file still to be restored in import (
// this run needs them. It works as follows: "context"
"fmt"
"sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/log"
)
// restoreSweeper frees cached blobs once all files that reference any of
// their chunks have been restored. It works as follows:
// //
// 1. After each chunk is restored, chunkRestored accumulates a running // 1. Callers add a file's ID to an in-memory restored set via
// fileRestored once the file is fully written to disk.
// 2. After each chunk is restored, chunkRestored accumulates a running
// byte count. // byte count.
// 2. When the accumulator crosses a threshold (one hundredth of the // 3. When the accumulator crosses a threshold (one hundredth of the
// configured blob size — so a sweep runs about a hundred times per // configured blob size — so a sweep runs about a hundred times per
// blob's worth of restored bytes), the sweeper iterates every key in // blob's worth of restored bytes), the sweeper iterates every key in
// the cache and deletes each blob that the restore plan no longer // the cache. For each cached blob it asks the DB which files
// needs. // reference any chunk in that blob, then compares that list against
// the in-memory restored set. If any referencing file is missing
// from the set the blob is kept; otherwise the cache entry is
// deleted.
// //
// The plan drops a file once it is restored or has failed, and holds // All DB reads happen against the snapshot's temporary metadata DB,
// only the files this run restores, so files outside the requested paths // which is local, indexed, and not under contention — the queries are
// never keep a blob. // cheap and run at most once per blob per sweep interval.
type restoreSweeper struct { type restoreSweeper struct {
plan *restorePlan ctx context.Context //nolint:containedctx // ctx bound at construction by design
repos *database.Repositories
cache *blobDiskCache cache *blobDiskCache
threshold int64 threshold int64
bytesAccum int64 bytesAccum int64
restored map[string]struct{}
} }
// newRestoreSweeper returns a sweeper that triggers eviction every // newRestoreSweeper returns a sweeper that triggers eviction every
// `threshold` bytes restored. Callers should pass blob_size_limit/100. // `threshold` bytes restored. Callers should pass blob_size_limit/100.
func newRestoreSweeper( func newRestoreSweeper(
plan *restorePlan, ctx context.Context,
repos *database.Repositories,
cache *blobDiskCache, cache *blobDiskCache,
threshold int64, threshold int64,
) *restoreSweeper { ) *restoreSweeper {
@@ -33,12 +49,21 @@ func newRestoreSweeper(
} }
return &restoreSweeper{ return &restoreSweeper{
plan: plan, ctx: ctx,
repos: repos,
cache: cache, cache: cache,
threshold: threshold, threshold: threshold,
restored: make(map[string]struct{}),
} }
} }
// fileRestored records a file as fully restored. After this call, any
// blob whose only remaining references come from files in the restored
// set may be evicted on the next sweep.
func (s *restoreSweeper) fileRestored(fileID string) {
s.restored[fileID] = struct{}{}
}
// chunkRestored accounts n bytes against the sweep threshold and runs a // chunkRestored accounts n bytes against the sweep threshold and runs a
// sweep if the threshold has been crossed since the last sweep. // sweep if the threshold has been crossed since the last sweep.
func (s *restoreSweeper) chunkRestored(n int64) { func (s *restoreSweeper) chunkRestored(n int64) {
@@ -51,11 +76,60 @@ func (s *restoreSweeper) chunkRestored(n int64) {
s.sweep() s.sweep()
} }
// sweep deletes every cached blob that no pending file references. // sweep deletes any cached blob whose chunks are no longer referenced
// by an unrestored file. Per-blob DB failures are logged and the blob
// is kept — we'd rather hold a blob longer than risk a re-download.
func (s *restoreSweeper) sweep() { func (s *restoreSweeper) sweep() {
for _, blobHash := range s.cache.Keys() { for _, blobHash := range s.cache.Keys() {
if !s.plan.pendingFileUsesBlob(blobHash) { needed, err := s.blobStillNeeded(blobHash)
if err != nil {
log.Debug("sweeper referencing-files query failed",
"blob_hash", blobHash[:16], "error", err)
continue
}
if !needed {
s.cache.Delete(blobHash) s.cache.Delete(blobHash)
} }
} }
} }
// blobStillNeeded returns true if any file that references a chunk in
// this blob has not yet been restored. On any error the function
// returns true — keeping the blob is always the safe answer because we
// can't prove we're done with it.
func (s *restoreSweeper) blobStillNeeded(blobHash string) (bool, error) {
rows, err := s.repos.DB().Conn().QueryContext(s.ctx, `
SELECT DISTINCT fc.file_id
FROM file_chunks fc
JOIN blob_chunks bc ON bc.chunk_hash = fc.chunk_hash
JOIN blobs b ON b.id = bc.blob_id
WHERE b.blob_hash = ?
`, blobHash)
if err != nil {
return true, fmt.Errorf("querying referencing files: %w", err)
}
defer func() { _ = rows.Close() }()
for rows.Next() {
var fileID string
err := rows.Scan(&fileID)
if err != nil {
return true, fmt.Errorf("scanning file_id: %w", err)
}
if _, ok := s.restored[fileID]; !ok {
return true, nil
}
}
err = rows.Err()
if err != nil {
return true, err
}
return false, nil
}
-163
View File
@@ -1,163 +0,0 @@
package vaultik //nolint:testpackage // reads the restore blob cache's peak size
import (
"context"
"fmt"
"io"
"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/log"
"sneak.berlin/go/vaultik/internal/storage"
"sneak.berlin/go/vaultik/internal/ui"
)
// TestRestoreFreesBlobsNoPendingFileNeeds asserts that the restore cache
// frees a blob once no file still to be restored in this run needs it.
// Every original file has a copy with the same content, so every blob
// holds chunks of copies: a restore that leaves the copies out of its
// paths, or fails on them, must not keep the blobs for them.
func TestRestoreFreesBlobsNoPendingFileNeeds(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
// Small files keep the backup and both restores quick.
const fileBytes = 16 * 1024
fs := afero.NewOsFs()
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "source")
originalsDir := filepath.Join(dataDir, "originals")
copiesDir := filepath.Join(dataDir, "copies")
storeDir := filepath.Join(tempDir, "remote")
dbPath := filepath.Join(tempDir, "index.sqlite")
originals := writeFilesWithCopies(t, fs, originalsDir, copiesDir, fileBytes)
ctx := context.Background()
cfg, storer, snapshotID := setupLocalityBackup(
ctx, t, fs, dataDir, storeDir, dbPath, fileBytes)
t.Run("restore by path", func(t *testing.T) {
t.Parallel()
restoreDir := filepath.Join(tempDir, "by-path")
peak, err := restorePeakCachedBlobs(ctx, t, fs, cfg, storer,
&RestoreOptions{
SnapshotID: snapshotID,
TargetDir: restoreDir,
Paths: []string{originalsDir},
})
require.NoError(t, err)
requireRestoredFiles(t, fs, restoreDir, originals)
assert.LessOrEqualf(t, peak, 1,
"peak cached blobs was %d; blobs were kept for unselected files",
peak)
})
t.Run("restore with failed files", func(t *testing.T) {
t.Parallel()
restoreDir := filepath.Join(tempDir, "with-failures")
// A regular file where the copies directory belongs makes every
// copy fail to restore.
blocker := filepath.Join(restoreDir, copiesDir)
require.NoError(t, fs.MkdirAll(filepath.Dir(blocker), 0o755))
require.NoError(t, afero.WriteFile(fs, blocker, nil, 0o644))
peak, err := restorePeakCachedBlobs(ctx, t, fs, cfg, storer,
&RestoreOptions{
SnapshotID: snapshotID,
TargetDir: restoreDir,
SkipErrors: true,
})
require.ErrorIs(t, err, errFilesFailedRestore)
requireRestoredFiles(t, fs, restoreDir, originals)
assert.LessOrEqualf(t, peak, 1,
"peak cached blobs was %d; blobs were kept for failed files", peak)
})
}
// writeFilesWithCopies writes 15 random files of fileBytes into
// originalsDir and a copy of each into copiesDir. With
// setupLocalityBackup given the same size, each file is one chunk and the
// 15 chunks fill 3 blobs. Returns the originals' content by path.
func writeFilesWithCopies(
t *testing.T, fs afero.Fs, originalsDir, copiesDir string, fileBytes int,
) map[string][]byte {
t.Helper()
const fileCount = 15
require.NoError(t, fs.MkdirAll(originalsDir, 0o755))
require.NoError(t, fs.MkdirAll(copiesDir, 0o755))
originals := make(map[string][]byte, fileCount)
for i := range fileCount {
name := fmt.Sprintf("file-%02d.bin", i)
data := randomBytes(t, fileBytes)
original := filepath.Join(originalsDir, name)
originals[original] = data
require.NoError(t, afero.WriteFile(fs, original, data, 0o644))
require.NoError(t, afero.WriteFile(
fs, filepath.Join(copiesDir, name), data, 0o644))
}
return originals
}
// requireRestoredFiles requires every file in want to have been restored
// under restoreDir with its content intact.
func requireRestoredFiles(
t *testing.T, fs afero.Fs, restoreDir string, want map[string][]byte,
) {
t.Helper()
for path, data := range want {
got, err := afero.ReadFile(fs, filepath.Join(restoreDir, path))
require.NoErrorf(t, err, "file missing after restore: %s", path)
require.Equalf(t, data, got, "byte mismatch for %s", path)
}
}
// restorePeakCachedBlobs runs a restore and returns the most blobs its
// cache held at once, together with the restore's error.
func restorePeakCachedBlobs(
ctx context.Context, t *testing.T, fs afero.Fs,
cfg *config.Config, storer storage.Storer, opts *RestoreOptions,
) (int, error) {
t.Helper()
var cache *blobDiskCache
v := &Vaultik{
Config: cfg,
Storage: storer,
Fs: fs,
Stdout: io.Discard,
Stderr: io.Discard,
UI: ui.NewWithColor(io.Discard, false),
restoreCacheObserver: func(c *blobDiskCache) {
cache = c
},
}
v.SetContext(ctx)
err := v.Restore(opts)
require.NotNil(t, cache, "restoreCacheObserver must fire during restore")
return cache.PeakLen(), err
}
+18 -40
View File
@@ -1228,11 +1228,9 @@ const snapshotRemoveCommandHint = "vaultik snapshot remove"
// destination store. Blobs are NOT touched: removing a snapshot's // destination store. Blobs are NOT touched: removing a snapshot's
// blobs requires reading every remaining remote manifest (the remote // blobs requires reading every remaining remote manifest (the remote
// may hold snapshots this host doesn't know about), which is what // may hold snapshots this host doesn't know about), which is what
// `vaultik prune` is for — when metadata was removed from the // `vaultik prune` is for — the command prints the prune invocation to
// destination store, the text output names the prune invocation to run // run as a follow-up. When the remote is unreachable the command still
// as a follow-up. When removing the metadata fails, a snapshot in the // completes the local-DB removal and warns.
// local index is still removed from it, with a warning; a snapshot
// only on the destination store is an error.
func (v *Vaultik) RemoveSnapshot( func (v *Vaultik) RemoveSnapshot(
snapshotID string, opts *RemoveOptions, snapshotID string, opts *RemoveOptions,
) (*RemoveResult, error) { ) (*RemoveResult, error) {
@@ -1267,10 +1265,8 @@ func (v *Vaultik) RemoveSnapshot(
} }
} }
result.RemoteRemoved, err = v.removeSnapshotRemote( if !opts.LocalOnly {
snapshotID, inLocalIndex, opts) result.RemoteRemoved = v.removeSnapshotRemote(snapshotID, opts)
if err != nil {
return result, err
} }
if v.SnapshotManager != nil { if v.SnapshotManager != nil {
@@ -1395,33 +1391,19 @@ func (v *Vaultik) confirmRemoveSnapshot(snapshotID string, opts *RemoveOptions)
} }
// removeSnapshotRemote strips the snapshot's metadata from the // removeSnapshotRemote strips the snapshot's metadata from the
// destination store. It returns true only when the delete finished // destination store, warning and proceeding on failure: the local-DB
// without an error and removed at least one object; under LocalOnly it // removal has already happened, so the user is told the remote half
// does nothing. When the snapshot is only on the destination store, a // didn't finish and to run `vaultik snapshot remove` for the snapshot
// failure is returned as an error, because that removal is all the // again once the destination store is reachable (`vaultik prune` never
// command does. When the snapshot was in the local index, its removal // removes snapshot metadata). Returns true when the remote removal
// there has already happened, so a failure is only a warning telling the // succeeded and deleted something.
// user to run `vaultik snapshot remove` for the snapshot again once the func (v *Vaultik) removeSnapshotRemote(snapshotID string, opts *RemoveOptions) bool {
// destination store is reachable (`vaultik prune` never removes snapshot
// metadata).
func (v *Vaultik) removeSnapshotRemote(
snapshotID string, inLocalIndex bool, opts *RemoveOptions,
) (bool, error) {
if opts.LocalOnly {
return false, nil
}
log.Info("Removing snapshot metadata from remote storage", log.Info("Removing snapshot metadata from remote storage",
"snapshot_id", snapshotID) "snapshot_id", snapshotID)
remoteKey := snapshot.RemoteSnapshotKey(snapshotID) remoteKey := snapshot.RemoteSnapshotKey(snapshotID)
deleted, err := v.deleteRemoteSnapshotByKey(remoteKey) deleted, err := v.deleteRemoteSnapshotByKey(remoteKey)
if err != nil && !inLocalIndex {
return false, fmt.Errorf(
"removing snapshot metadata from destination store: %w", err)
}
if err != nil { if err != nil {
log.Warn("Could not remove snapshot metadata from remote storage; "+ log.Warn("Could not remove snapshot metadata from remote storage; "+
"run '"+snapshotRemoveCommandHint+"' with the snapshot's ID "+ "run '"+snapshotRemoveCommandHint+"' with the snapshot's ID "+
@@ -1436,10 +1418,10 @@ func (v *Vaultik) removeSnapshotRemote(
err, snapshotRemoveCommandHint, snapshotID) err, snapshotRemoveCommandHint, snapshotID)
} }
return false, nil return false
} }
return deleted > 0, nil return deleted > 0
} }
// RemoveAllSnapshots removes every snapshot known to the local // RemoveAllSnapshots removes every snapshot known to the local
@@ -1920,17 +1902,13 @@ func (v *Vaultik) PruneDatabase() (*PruneResult, error) {
snapshotsRemain := countDiff(snapshotCountBefore, &result.SnapshotsDeleted) snapshotsRemain := countDiff(snapshotCountBefore, &result.SnapshotsDeleted)
v.UI.Completef("Pruned local index database.") v.UI.Completef("Pruned local index database.")
v.UI.Detailf("Incomplete snapshots: %s removed "+ v.UI.Detailf("Incomplete snapshots: %s removed (%s remain).",
"(total snapshots left in the local index: %s).",
countText(&result.SnapshotsDeleted), countText(snapshotsRemain)) countText(&result.SnapshotsDeleted), countText(snapshotsRemain))
v.UI.Detailf("Orphaned files: %s removed "+ v.UI.Detailf("Orphaned files: %s removed (%s remain).",
"(total files left in the local index: %s).",
countText(result.FilesDeleted), countText(fileCountAfter)) countText(result.FilesDeleted), countText(fileCountAfter))
v.UI.Detailf("Orphaned chunks: %s removed "+ v.UI.Detailf("Orphaned chunks: %s removed (%s remain).",
"(total chunks left in the local index: %s).",
countText(result.ChunksDeleted), countText(chunkCountAfter)) countText(result.ChunksDeleted), countText(chunkCountAfter))
v.UI.Detailf("Orphaned blobs: %s removed "+ v.UI.Detailf("Orphaned blobs: %s removed (%s remain).",
"(total blobs left in the local index: %s).",
countText(result.BlobsDeleted), countText(blobCountAfter)) countText(result.BlobsDeleted), countText(blobCountAfter))
return result, nil return result, nil