Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ad7941c215 | ||
|
|
94412d9ef3 |
@@ -401,7 +401,9 @@ orphaned blob figures are reported as unknown; `--json` gives them as
|
|||||||
`null`, lists the remote key of each unreadable manifest in
|
`null`, lists the remote key of each unreadable manifest in
|
||||||
`unreadable_manifests` and counts the manifests under skipped names in
|
`unreadable_manifests` and counts the manifests under skipped names in
|
||||||
`skipped_manifest_count`. An unreadable manifest also leaves its
|
`skipped_manifest_count`. An unreadable manifest also leaves its
|
||||||
snapshot's blob count and blob size unknown, `null` in `--json`.
|
snapshot's blob count and blob size unknown, `null` in `--json`. A file
|
||||||
|
under `blobs/` whose name is not a blob hash is not counted as a blob,
|
||||||
|
since `prune` never deletes it.
|
||||||
* `--json`: Output as JSON
|
* `--json`: Output as JSON
|
||||||
|
|
||||||
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
**`remote nuke`**: Delete every snapshot's metadata and every blob from the
|
||||||
|
|||||||
@@ -22,15 +22,24 @@ the tag exists and is exercised; what is left is merging `next` to
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
- 2026-10-13: Made Ctrl-C or SIGTERM during a local index write end the
|
- 2026-10-13: Made a `file://` destination list through a symlinked
|
||||||
command as interrupted
|
directory ([issue #296](https://git.eeqj.de/sneak/vaultik/issues/296)).
|
||||||
([issue #301](https://git.eeqj.de/sneak/vaultik/issues/301)). Once the
|
A `metadata/` or `blobs/` directory that was a symlink to a directory
|
||||||
run is cancelled, `database/sql` rolls an open transaction back itself,
|
elsewhere was listed as one file, so the destination store listed as
|
||||||
so the rollback in `WithTx` failed with `sql.ErrTxDone` and the process
|
empty while backups and restores still worked through the link.
|
||||||
exited 1 with `FATAL: failed to rollback transaction`, skipping the
|
`prune` then removed the local snapshot records and deleted every blob,
|
||||||
interrupted exit and message. That result is no longer an error, and a
|
and `remote info` and `remote nuke` found no blobs. The listings now
|
||||||
rollback that does fail is returned with the transaction's error
|
read through the link as `Get` and `Stat` do. A link whose target is
|
||||||
instead of ending the process.
|
missing, such as one to a volume that is not mounted, or a link
|
||||||
|
pointing back up the tree makes them fail.
|
||||||
|
|
||||||
|
- 2026-10-13: Stopped `remote info` counting a file under `blobs/` that
|
||||||
|
is not a blob as an orphaned blob
|
||||||
|
([issue #295](https://git.eeqj.de/sneak/vaultik/issues/295)). A stray
|
||||||
|
file such as `.DS_Store` was reported as an orphaned blob and the
|
||||||
|
report advised running `prune`, which skips such a file and deleted
|
||||||
|
nothing. `remote info` now takes its blob list from the same listing
|
||||||
|
`prune` uses, in the table and in `--json`.
|
||||||
|
|
||||||
- 2026-10-13: Made `snapshot remove` fail for a snapshot that does not
|
- 2026-10-13: Made `snapshot remove` fail for a snapshot that does not
|
||||||
exist ([issue #293](https://git.eeqj.de/sneak/vaultik/issues/293)).
|
exist ([issue #293](https://git.eeqj.de/sneak/vaultik/issues/293)).
|
||||||
|
|||||||
@@ -3,7 +3,6 @@ package database
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -61,24 +60,22 @@ func (r *Repositories) WithTx(ctx context.Context, fn TxFunc) error {
|
|||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if p := recover(); p != nil {
|
if p := recover(); p != nil {
|
||||||
// The panic is the failure to report, so a rollback error
|
rollbackErr := tx.Rollback()
|
||||||
// is dropped.
|
if rollbackErr != nil {
|
||||||
_ = tx.Rollback()
|
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
||||||
|
}
|
||||||
|
|
||||||
panic(p)
|
panic(p)
|
||||||
|
} else if err != nil {
|
||||||
|
rollbackErr := tx.Rollback()
|
||||||
|
if rollbackErr != nil {
|
||||||
|
Fatalf("failed to rollback transaction: %v", rollbackErr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
err = fn(ctx, tx)
|
err = fn(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// Once ctx is cancelled, database/sql rolls the transaction
|
|
||||||
// back itself and Rollback returns sql.ErrTxDone.
|
|
||||||
rollbackErr := tx.Rollback()
|
|
||||||
if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
|
|
||||||
return fmt.Errorf("%w; rolling back transaction: %w",
|
|
||||||
err, rollbackErr)
|
|
||||||
}
|
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -235,64 +235,6 @@ func TestRepositoriesTransactionRollback(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// When the context is cancelled during the body, database/sql rolls the
|
|
||||||
// transaction back itself, so WithTx's own rollback finds it already
|
|
||||||
// done. WithTx must still return the body's error.
|
|
||||||
func TestRepositoriesTransactionContextCancelled(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db, cleanup := setupTestDB(t)
|
|
||||||
defer cleanup()
|
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
|
|
||||||
repos := database.NewRepositories(db)
|
|
||||||
|
|
||||||
err := repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
|
||||||
file := &database.File{
|
|
||||||
Path: "/test/cancelled_file.txt",
|
|
||||||
MTime: time.Now().Truncate(time.Second),
|
|
||||||
Size: 1024,
|
|
||||||
Mode: 0644,
|
|
||||||
UID: 1000,
|
|
||||||
GID: 1000,
|
|
||||||
}
|
|
||||||
|
|
||||||
err := repos.Files.Create(ctx, tx, file)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
cancel()
|
|
||||||
|
|
||||||
// database/sql rolls back from a goroutine of its own; wait
|
|
||||||
// until it has, so the test does not depend on that race.
|
|
||||||
for {
|
|
||||||
_, err = tx.ExecContext(context.WithoutCancel(ctx), "SELECT 1")
|
|
||||||
if errors.Is(err, sql.ErrTxDone) {
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
|
|
||||||
time.Sleep(time.Millisecond)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
if !errors.Is(err, context.Canceled) {
|
|
||||||
t.Fatalf("expected context.Canceled, got: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
file, err := repos.Files.GetByPath(
|
|
||||||
context.Background(), "/test/cancelled_file.txt",
|
|
||||||
)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("error checking for file: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if file != nil {
|
|
||||||
t.Error("file should not exist after the cancelled transaction")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestRepositoriesReadTransaction(t *testing.T) {
|
func TestRepositoriesReadTransaction(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
+45
-10
@@ -120,9 +120,10 @@ func (f *FileStorer) Delete(_ context.Context, key string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// List returns all keys with the given prefix. It fails when the
|
// List returns all keys with the given prefix, reading through symlinked
|
||||||
// destination directory is missing; a missing prefix under it is an
|
// directories. It fails when the destination directory is missing or the
|
||||||
// empty listing.
|
// prefix is a symlink whose target is missing; a missing prefix under the
|
||||||
|
// destination directory is an empty listing.
|
||||||
func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
|
func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
|
||||||
var keys []string
|
var keys []string
|
||||||
|
|
||||||
@@ -133,8 +134,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
// Check if the prefix exists
|
exists, err := f.prefixExists(basePath)
|
||||||
exists, err := afero.Exists(f.fs, basePath)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("checking path: %w", err)
|
return nil, fmt.Errorf("checking path: %w", err)
|
||||||
}
|
}
|
||||||
@@ -143,7 +143,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
return keys, nil // Empty list for non-existent prefix
|
return keys, nil // Empty list for non-existent prefix
|
||||||
}
|
}
|
||||||
|
|
||||||
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
err = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -176,7 +176,9 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ListStream returns a channel of ObjectInfo for large result sets. Like
|
// ListStream returns a channel of ObjectInfo for large result sets. Like
|
||||||
// List, it sends an error when the destination directory is missing.
|
// List, it reads through symlinked directories and sends an error when
|
||||||
|
// the destination directory is missing or the prefix is a symlink whose
|
||||||
|
// target is missing.
|
||||||
func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan ObjectInfo {
|
func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan ObjectInfo {
|
||||||
ch := make(chan ObjectInfo)
|
ch := make(chan ObjectInfo)
|
||||||
|
|
||||||
@@ -192,8 +194,7 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
// Check if the prefix exists
|
exists, err := f.prefixExists(basePath)
|
||||||
exists, err := afero.Exists(f.fs, basePath)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
ch <- ObjectInfo{Err: fmt.Errorf("checking path: %w", err)}
|
ch <- ObjectInfo{Err: fmt.Errorf("checking path: %w", err)}
|
||||||
|
|
||||||
@@ -204,7 +205,7 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
return // Empty channel for non-existent prefix
|
return // Empty channel for non-existent prefix
|
||||||
}
|
}
|
||||||
|
|
||||||
_ = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
_ = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
// Check context cancellation
|
// Check context cancellation
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
@@ -352,6 +353,40 @@ func (f *FileStorer) writeAtomic(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// walk is afero.Walk, except that it reads through a symlinked directory
|
||||||
|
// the way Get and Stat do, where afero.Walk reports the link as one file.
|
||||||
|
// afero.Walk uses Lstat, which does not follow a link, whenever the
|
||||||
|
// filesystem has one; the struct below carries only the afero.Fs methods,
|
||||||
|
// so the walk uses Stat instead. A link back up the tree is followed until
|
||||||
|
// the operating system refuses the path for too many levels of links, and
|
||||||
|
// the walk then fails.
|
||||||
|
func (f *FileStorer) walk(root string, fn filepath.WalkFunc) error {
|
||||||
|
return afero.Walk(struct{ afero.Fs }{f.fs}, root, fn)
|
||||||
|
}
|
||||||
|
|
||||||
|
// prefixExists is afero.Exists, except that it does not follow a symlink
|
||||||
|
// at path. A symlinked prefix whose target is missing, such as a link to
|
||||||
|
// a volume that is not mounted, therefore exists, and walk then fails on
|
||||||
|
// it; with afero.Exists it would list as empty, and prune would delete
|
||||||
|
// every blob.
|
||||||
|
func (f *FileStorer) prefixExists(path string) (bool, error) {
|
||||||
|
lstater, ok := f.fs.(afero.Lstater)
|
||||||
|
if !ok {
|
||||||
|
return afero.Exists(f.fs, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, _, err := lstater.LstatIfPossible(path)
|
||||||
|
if err == nil {
|
||||||
|
return true, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if os.IsNotExist(err) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
|
||||||
// fullPath returns the full filesystem path for a key.
|
// fullPath returns the full filesystem path for a key.
|
||||||
func (f *FileStorer) fullPath(key string) string {
|
func (f *FileStorer) fullPath(key string) string {
|
||||||
return filepath.Join(f.basePath, key)
|
return filepath.Join(f.basePath, key)
|
||||||
|
|||||||
@@ -4,7 +4,10 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"reflect"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"sneak.berlin/go/vaultik/internal/storage"
|
"sneak.berlin/go/vaultik/internal/storage"
|
||||||
@@ -60,3 +63,151 @@ func TestFileStorerListMissingDestination(t *testing.T) {
|
|||||||
t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
|
t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestFileStorerListsThroughSymlinkedDirectory checks that List and
|
||||||
|
// ListStream read through a symlinked metadata/ or blobs/ directory, as Get
|
||||||
|
// and Stat do, instead of listing the link as one key. Listing the whole
|
||||||
|
// destination meets the link part-way through the walk; listing its prefix
|
||||||
|
// starts the walk at the link.
|
||||||
|
func TestFileStorerListsThroughSymlinkedDirectory(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, dir := range []string{"metadata", "blobs"} {
|
||||||
|
t.Run(dir, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
base := t.TempDir()
|
||||||
|
key := dir + "/aa/bb/object"
|
||||||
|
|
||||||
|
s, err := storage.NewFileStorer(base)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewFileStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.Put(context.Background(), key, strings.NewReader("data"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
linkPath := filepath.Join(base, dir)
|
||||||
|
movedPath := filepath.Join(t.TempDir(), dir)
|
||||||
|
|
||||||
|
err = os.Rename(linkPath, movedPath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("moving %s away: %v", dir, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = os.Symlink(movedPath, linkPath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("linking %s: %v", dir, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
want := []string{key}
|
||||||
|
|
||||||
|
for _, prefix := range []string{"", dir + "/"} {
|
||||||
|
got := listSorted(t, s, prefix)
|
||||||
|
if !reflect.DeepEqual(got, want) {
|
||||||
|
t.Errorf("List(%q) = %v, want %v", prefix, got, want)
|
||||||
|
}
|
||||||
|
|
||||||
|
got = listStreamKeys(t, s, prefix)
|
||||||
|
if !reflect.DeepEqual(got, want) {
|
||||||
|
t.Errorf("ListStream(%q) = %v, want %v", prefix, got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFileStorerListFailsOnSymlinkWithMissingTarget checks that List and
|
||||||
|
// ListStream fail on a symlinked metadata/ or blobs/ directory whose target
|
||||||
|
// is missing, such as a link to a volume that is not mounted, as they do on
|
||||||
|
// a missing destination directory, instead of listing it as empty.
|
||||||
|
func TestFileStorerListFailsOnSymlinkWithMissingTarget(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
for _, dir := range []string{"metadata", "blobs"} {
|
||||||
|
t.Run(dir, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
base := t.TempDir()
|
||||||
|
|
||||||
|
s, err := storage.NewFileStorer(base)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewFileStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
unmounted := filepath.Join(t.TempDir(), "unmounted")
|
||||||
|
|
||||||
|
err = os.Symlink(unmounted, filepath.Join(base, dir))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Symlink: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, prefix := range []string{"", dir + "/"} {
|
||||||
|
keys, err := s.List(ctx, prefix)
|
||||||
|
if !errors.Is(err, fs.ErrNotExist) {
|
||||||
|
t.Errorf("List(%q) = %v, %v; want a not-exist error",
|
||||||
|
prefix, keys, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var streamErr error
|
||||||
|
|
||||||
|
for object := range s.ListStream(ctx, prefix) {
|
||||||
|
if object.Err != nil {
|
||||||
|
streamErr = object.Err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if !errors.Is(streamErr, fs.ErrNotExist) {
|
||||||
|
t.Errorf("ListStream(%q) error = %v, want a not-exist error",
|
||||||
|
prefix, streamErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFileStorerListFailsOnSymlinkLoop checks that a symlink pointing back
|
||||||
|
// at the destination makes List and ListStream fail instead of walking it
|
||||||
|
// forever.
|
||||||
|
func TestFileStorerListFailsOnSymlinkLoop(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
base := t.TempDir()
|
||||||
|
|
||||||
|
s, err := storage.NewFileStorer(base)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewFileStorer: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = s.Put(ctx, testBlobKey, strings.NewReader("data"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = os.Symlink(base, filepath.Join(base, "blobs", "loop"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Symlink: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
keys, err := s.List(ctx, "blobs/")
|
||||||
|
if err == nil {
|
||||||
|
t.Errorf("List = %d keys and no error, want an error", len(keys))
|
||||||
|
}
|
||||||
|
|
||||||
|
var streamErr error
|
||||||
|
|
||||||
|
for object := range s.ListStream(ctx, "blobs/") {
|
||||||
|
if object.Err != nil {
|
||||||
|
streamErr = object.Err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if streamErr == nil {
|
||||||
|
t.Error("ListStream sent no error, want one")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -167,10 +167,6 @@ const (
|
|||||||
// metadataKeyParts is the minimum "/"-separated segment count of a
|
// metadataKeyParts is the minimum "/"-separated segment count of a
|
||||||
// metadata object key (metadata/<snapshot-id>/<filename>).
|
// metadata object key (metadata/<snapshot-id>/<filename>).
|
||||||
metadataKeyParts = 3
|
metadataKeyParts = 3
|
||||||
|
|
||||||
// blobKeyParts is the minimum "/"-separated segment count of a blob
|
|
||||||
// object key (blobs/<aa>/<bb>/<hash>).
|
|
||||||
blobKeyParts = 4
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// SnapshotMetadataInfo contains information about a single snapshot's metadata
|
// SnapshotMetadataInfo contains information about a single snapshot's metadata
|
||||||
@@ -454,23 +450,16 @@ func (v *Vaultik) scanRemoteBlobStorage(
|
|||||||
v.stdoutf("Scanning blobs...\n")
|
v.stdoutf("Scanning blobs...\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
blobCh := v.Storage.ListStream(v.ctx, "blobs/")
|
// Use prune's listing, so a file under blobs/ that prune would never
|
||||||
allBlobs := make(map[string]int64)
|
// delete is not counted as a blob, and so not as an orphaned one.
|
||||||
|
allBlobs, err := v.listAllRemoteBlobs()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
for obj := range blobCh {
|
for _, size := range allBlobs {
|
||||||
if obj.Err != nil {
|
|
||||||
return fmt.Errorf("listing blobs: %w", obj.Err)
|
|
||||||
}
|
|
||||||
|
|
||||||
parts := strings.Split(obj.Key, "/")
|
|
||||||
if len(parts) < blobKeyParts {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
hash := parts[3]
|
|
||||||
allBlobs[hash] = obj.Size
|
|
||||||
result.TotalBlobCount++
|
result.TotalBlobCount++
|
||||||
result.TotalBlobSize += obj.Size
|
result.TotalBlobSize += size
|
||||||
}
|
}
|
||||||
|
|
||||||
// A blob named only by a manifest that could not be read, or by one
|
// A blob named only by a manifest that could not be read, or by one
|
||||||
|
|||||||
@@ -23,6 +23,10 @@ var errNukeRequiresForce = errors.New(
|
|||||||
// metadata.
|
// metadata.
|
||||||
const metadataDirName = "metadata"
|
const metadataDirName = "metadata"
|
||||||
|
|
||||||
|
// blobKeyParts is the "/"-separated segment count of a blob object key
|
||||||
|
// (blobs/<aa>/<bb>/<hash>).
|
||||||
|
const blobKeyParts = 4
|
||||||
|
|
||||||
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
// NukeRemote deletes every snapshot's metadata and every blob from remote
|
||||||
// storage, along with any object an upload cut off part-way left under a
|
// storage, along with any object an upload cut off part-way left under a
|
||||||
// temporary `.partial` name. After this returns successfully the bucket
|
// temporary `.partial` name. After this returns successfully the bucket
|
||||||
|
|||||||
@@ -185,6 +185,39 @@ func TestRemoteInfo_SkipsNonConformingMetadataName(t *testing.T) {
|
|||||||
assert.NotContains(t, doc, "unreadable_manifests")
|
assert.NotContains(t, doc, "unreadable_manifests")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRemoteInfo_DoesNotCountNonBlobFileUnderBlobs checks that a file
|
||||||
|
// under blobs/ whose name is not a blob hash is not counted as a blob.
|
||||||
|
// prune never deletes such a file, so counting it as an orphaned blob
|
||||||
|
// would advise a prune that removes nothing.
|
||||||
|
func TestRemoteInfo_DoesNotCountNonBlobFileUnderBlobs(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newListEnv(t)
|
||||||
|
env.addRemote(t, listRemoteID, time.Date(2026, 3, 2, 0, 0, 0, 0, time.UTC))
|
||||||
|
addBlob(t, env.store.testStorer, testBlobHashA)
|
||||||
|
require.NoError(t, env.store.Put(context.Background(),
|
||||||
|
"blobs/"+testBlobHashA[:2]+"/"+testBlobHashA[2:4]+"/.DS_Store",
|
||||||
|
bytes.NewReader([]byte("junk"))))
|
||||||
|
|
||||||
|
require.NoError(t, env.v.RemoteInfo(false))
|
||||||
|
|
||||||
|
text := env.stdout.String()
|
||||||
|
assert.Contains(t, text, "Total blobs on remote: 1 (")
|
||||||
|
assert.Contains(t, text, "Orphaned (unreferenced): 0 (")
|
||||||
|
assert.NotContains(t, text, "vaultik prune")
|
||||||
|
|
||||||
|
env.stdout.Reset()
|
||||||
|
require.NoError(t, env.v.RemoteInfo(true))
|
||||||
|
|
||||||
|
var doc map[string]any
|
||||||
|
|
||||||
|
require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &doc))
|
||||||
|
assert.InDelta(t, 1, doc["total_blob_count"], 0)
|
||||||
|
assert.InDelta(t, 0, doc["orphaned_blob_count"], 0)
|
||||||
|
assert.InDelta(t, 0, doc["orphaned_blob_size"], 0)
|
||||||
|
}
|
||||||
|
|
||||||
// TestRemoteInfo_DirectoryWithoutManifestLeavesOrphansKnown checks that
|
// TestRemoteInfo_DirectoryWithoutManifestLeavesOrphansKnown checks that
|
||||||
// a directory under metadata/ holding no manifest.json.zst, such as one
|
// a directory under metadata/ holding no manifest.json.zst, such as one
|
||||||
// left by a backup interrupted before its manifest upload, leaves the
|
// left by a backup interrupted before its manifest upload, leaves the
|
||||||
|
|||||||
@@ -0,0 +1,121 @@
|
|||||||
|
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/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestPruneDeletesNothingThroughSymlinkedDirectory checks that prune keeps
|
||||||
|
// every blob and the local snapshot record when the metadata/ or blobs/
|
||||||
|
// directory of a file:// destination is a symlink to a directory elsewhere
|
||||||
|
// (https://git.eeqj.de/sneak/vaultik/issues/296). A backup writes through
|
||||||
|
// such a link, so prune has to list through it too.
|
||||||
|
func TestPruneDeletesNothingThroughSymlinkedDirectory(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, dir := range []string{"metadata", "blobs"} {
|
||||||
|
t.Run(dir, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
blobFiles := blobFilesIn(t, storeDir)
|
||||||
|
moveBehindSymlink(t, storeDir, dir)
|
||||||
|
|
||||||
|
require.NoError(t, v.Prune(&vaultik.PruneOptions{Force: true}))
|
||||||
|
|
||||||
|
// When blobs/ is the link, these paths run through it.
|
||||||
|
for _, blobFile := range blobFiles {
|
||||||
|
assert.FileExists(t, blobFile, "prune must delete no blob")
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPruneFailsOnSymlinkedDirectoryWithMissingTarget checks that prune
|
||||||
|
// fails, deleting no blob and keeping the local snapshot record, when the
|
||||||
|
// metadata/ or blobs/ directory of a file:// destination is a symlink
|
||||||
|
// whose target is missing, as a link to a volume that is not mounted is.
|
||||||
|
func TestPruneFailsOnSymlinkedDirectoryWithMissingTarget(t *testing.T) {
|
||||||
|
log.Initialize(log.Config{})
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, dir := range []string{"metadata", "blobs"} {
|
||||||
|
t.Run(dir, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
storeDir := filepath.Join(t.TempDir(), "store")
|
||||||
|
v, repos, _ := backUpToFileDestination(ctx, t, storeDir)
|
||||||
|
blobFiles := blobFilesIn(t, storeDir)
|
||||||
|
targetPath := moveBehindSymlink(t, storeDir, dir)
|
||||||
|
|
||||||
|
// Unmount the volume the link points to, and mount it again
|
||||||
|
// once prune has run.
|
||||||
|
unmountedPath := targetPath + "-unmounted"
|
||||||
|
require.NoError(t, os.Rename(targetPath, unmountedPath))
|
||||||
|
|
||||||
|
err := v.Prune(&vaultik.PruneOptions{Force: true})
|
||||||
|
require.ErrorIs(t, err, fs.ErrNotExist)
|
||||||
|
|
||||||
|
require.NoError(t, os.Rename(unmountedPath, targetPath))
|
||||||
|
|
||||||
|
// When blobs/ is the link, these paths run through it.
|
||||||
|
for _, blobFile := range blobFiles {
|
||||||
|
assert.FileExists(t, blobFile, "prune must delete no blob")
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshots, err := repos.Snapshots.ListRecent(ctx, listRecentTestLimit)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Len(t, snapshots, 1, "prune must delete no local snapshot record")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// blobFilesIn returns the paths of the blob files under storeDir/blobs, and
|
||||||
|
// fails the test when there are none.
|
||||||
|
func blobFilesIn(t *testing.T, storeDir string) []string {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var blobFiles []string
|
||||||
|
|
||||||
|
err := filepath.WalkDir(filepath.Join(storeDir, "blobs"),
|
||||||
|
func(path string, entry fs.DirEntry, err error) error {
|
||||||
|
if err == nil && !entry.IsDir() {
|
||||||
|
blobFiles = append(blobFiles, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
return err
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotEmpty(t, blobFiles)
|
||||||
|
|
||||||
|
return blobFiles
|
||||||
|
}
|
||||||
|
|
||||||
|
// moveBehindSymlink moves storeDir/dir to a directory elsewhere, puts a
|
||||||
|
// symlink to it in its place, and returns where it went.
|
||||||
|
func moveBehindSymlink(t *testing.T, storeDir, dir string) string {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
linkPath := filepath.Join(storeDir, dir)
|
||||||
|
targetPath := filepath.Join(t.TempDir(), dir)
|
||||||
|
require.NoError(t, os.Rename(linkPath, targetPath))
|
||||||
|
require.NoError(t, os.Symlink(targetPath, linkPath))
|
||||||
|
|
||||||
|
return targetPath
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user