List a file:// destination through a symlinked directory (closes #296)

List and ListStream walked the destination with afero.Walk, which uses
Lstat and so reported a symlinked metadata/ or blobs/ directory as one
file, while Put, Get and Stat follow the link. The store then listed as
empty, and prune removed the local snapshot records and deleted every
blob. The listings, and DeletePartialUploads, which remote nuke uses,
now walk a view of the filesystem without its Lstat method, so afero
falls back to Stat and enters the link. A prefix lists as empty only
when it is truly missing: when it or a directory above it is a symlink
whose target is missing, such as a link to an unmounted volume, the
listing fails.

Judgement call: a symlink loop is not detected; it is followed until the
system refuses the path, and the listing fails.

Model: opus-5-5

internal/storage/file.go

@@ -120,9 +120,10 @@ func (f *FileStorer) Delete(_ context.Context, key string) error {
 	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
-// empty listing.
+// when the prefix or a directory above it is a symlink whose target is
+// missing; a prefix that is truly missing is an empty listing.
 func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error) {
 	var keys []string
 
@@ -133,8 +134,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
 
 	basePath := f.fullPath(prefix)
 
-	// Check if the prefix exists
+	exists, err := f.prefixExists(prefix)
-	exists, err := afero.Exists(f.fs, basePath)
 	if err != nil {
 		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
 	}
 
-	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 {
 			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
-// 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 when the prefix or a directory
+// above it is a symlink whose target is missing.
 func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-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)
 
-		// Check if the prefix exists
+		exists, err := f.prefixExists(prefix)
-		exists, err := afero.Exists(f.fs, basePath)
 		if err != nil {
 			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
 		}
 
-		_ = 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
 			select {
 			case <-ctx.Done():
@@ -243,11 +244,12 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
 }
 
 // DeletePartialUploads removes every file under prefix whose name ends in
-// tempSuffix. A missing prefix has none to remove.
+// tempSuffix. Like List, it reads through symlinked directories and fails
+// on one whose target is missing. A missing prefix has none to remove.
 func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) error {
 	basePath := f.fullPath(prefix)
 
-	exists, err := afero.Exists(f.fs, basePath)
+	exists, err := f.prefixExists(prefix)
 	if err != nil {
 		return fmt.Errorf("checking path: %w", err)
 	}
@@ -256,7 +258,7 @@ func (f *FileStorer) DeletePartialUploads(ctx context.Context, prefix string) er
 		return nil
 	}
 
-	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 {
 			return err
 		}
@@ -352,6 +354,54 @@ func (f *FileStorer) writeAtomic(
 	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 reports whether prefix exists under the destination
+// directory. It checks each directory on the way down to prefix in turn,
+// and the first one Stat cannot find decides: if it is not there at all,
+// prefix is missing; if it is a symlink whose target is missing, such as
+// a link to a volume that is not mounted, prefixExists fails, since
+// listing it as empty would let prune delete every blob.
+func (f *FileStorer) prefixExists(prefix string) (bool, error) {
+	lstater, canLstat := f.fs.(afero.Lstater)
+	path := f.basePath
+
+	for name := range strings.SplitSeq(prefix, "/") {
+		path = filepath.Join(path, name)
+
+		_, err := f.fs.Stat(path)
+		if err == nil {
+			continue
+		}
+
+		if !os.IsNotExist(err) {
+			return false, err
+		}
+
+		// Lstat does not follow a symlink at path, so it finds the link
+		// that Stat could not follow.
+		if canLstat {
+			_, _, lstatErr := lstater.LstatIfPossible(path)
+			if lstatErr == nil {
+				return false, err
+			}
+		}
+
+		return false, nil
+	}
+
+	return true, nil
+}
+
 // fullPath returns the full filesystem path for a key.
 func (f *FileStorer) fullPath(key string) string {
 	return filepath.Join(f.basePath, key)

internal/storage/file_backend_test.go

@@ -4,7 +4,10 @@ import (
 	"context"
 	"errors"
 	"io/fs"
+	"os"
 	"path/filepath"
+	"reflect"
+	"strings"
 	"testing"
 
 	"sneak.berlin/go/vaultik/internal/storage"
@@ -60,3 +63,225 @@ func TestFileStorerListMissingDestination(t *testing.T) {
 		t.Errorf("ListStream error = %v, want a not-exist error", streamErr)
 	}
 }
+
+// moveBehindSymlink moves base/dir to a directory elsewhere and puts a
+// symlink to it in its place.
+func moveBehindSymlink(t *testing.T, base, dir string) {
+	t.Helper()
+
+	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)
+	}
+}
+
+// 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. A prefix missing behind the link lists as
+// empty.
+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)
+			}
+
+			moveBehindSymlink(t, base, dir)
+
+			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)
+				}
+			}
+
+			absent := dir + "/absent/"
+
+			if got := listSorted(t, s, absent); len(got) != 0 {
+				t.Errorf("List(%q) = %v, want empty", absent, got)
+			}
+
+			if got := listStreamKeys(t, s, absent); len(got) != 0 {
+				t.Errorf("ListStream(%q) = %v, want empty", absent, got)
+			}
+		})
+	}
+}
+
+// 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. That
+// holds for a prefix below the link too.
+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 + "/", dir + "/aa/"} {
+				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")
+	}
+}
+
+// TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory checks that
+// DeletePartialUploads, which remote nuke uses, removes a leftover temp
+// file behind a symlinked metadata/ or blobs/ directory and keeps the
+// object at the real key.
+func TestFileStorerDeletePartialUploadsThroughSymlinkedDirectory(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()
+			key := dir + "/aa/bb/object"
+
+			s, err := storage.NewFileStorer(base)
+			if err != nil {
+				t.Fatalf("NewFileStorer: %v", err)
+			}
+
+			err = s.Put(ctx, key, strings.NewReader("data"))
+			if err != nil {
+				t.Fatalf("Put: %v", err)
+			}
+
+			leftover := filepath.Join(base, key+"-123456.partial")
+
+			err = os.WriteFile(leftover, []byte("half"), 0o600)
+			if err != nil {
+				t.Fatalf("writing leftover temp file: %v", err)
+			}
+
+			moveBehindSymlink(t, base, dir)
+
+			err = s.DeletePartialUploads(ctx, dir+"/")
+			if err != nil {
+				t.Fatalf("DeletePartialUploads: %v", err)
+			}
+
+			// leftover runs through the link.
+			_, err = os.Stat(leftover)
+			if !os.IsNotExist(err) {
+				t.Errorf("leftover temp file was not removed: %v", err)
+			}
+
+			_, err = s.Stat(ctx, key)
+			if err != nil {
+				t.Errorf("Stat of the real key: %v", err)
+			}
+		})
+	}
+}

internal/vaultik/destination_validation_test.go

@@ -161,13 +161,14 @@ func TestVerifyManifestBlobsRejectsShortHash(t *testing.T) {
 		Blobs: []snapshot.BlobInfo{{Hash: "abc", CompressedSize: 1}},
 	}
 
-	verified, missing, mismatched, missingSize, err :=
+	verified, missing, mismatched, missingSize, unchecked, err :=
 		v.verifyManifestBlobs(manifest, &VerifyOptions{JSON: true})
 	require.ErrorIs(t, err, errInvalidBlobHash)
 	require.Zero(t, verified)
 	require.Zero(t, missing)
 	require.Zero(t, mismatched)
 	require.Zero(t, missingSize)
+	require.Empty(t, unchecked)
 }
 
 // TestVerifyBlobChunksRejectsNegativeLength proves a blob_chunks row with a

internal/vaultik/shallow_verify_test.go

@@ -4,15 +4,20 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
 	"os"
+	"path"
 	"path/filepath"
+	"slices"
 	"testing"
 
 	"github.com/spf13/afero"
 	"github.com/stretchr/testify/require"
 	"sneak.berlin/go/vaultik/internal/log"
 	"sneak.berlin/go/vaultik/internal/snapshot"
+	"sneak.berlin/go/vaultik/internal/storage"
 	"sneak.berlin/go/vaultik/internal/ui"
 	"sneak.berlin/go/vaultik/internal/vaultik"
 )
@@ -151,6 +156,224 @@ func TestShallowVerifyDetectsMissingDatabase(t *testing.T) {
 		"shallow verify must fail when db.zst.age is absent")
 }
 
+// errStatDenied stands in for any storage error other than not-found: a
+// permission error, an S3 403, a dropped connection.
+var errStatDenied = errors.New("injected stat failure")
+
+// statFailingStorer fails Stat with errStatDenied for the keys in
+// failKeys and passes every other call to the wrapped Storer.
+type statFailingStorer struct {
+	storage.Storer
+
+	failKeys []string
+}
+
+func (s statFailingStorer) Stat(
+	ctx context.Context, key string,
+) (*storage.ObjectInfo, error) {
+	if slices.Contains(s.failKeys, key) {
+		return nil, fmt.Errorf("stat %q: %w", key, errStatDenied)
+	}
+
+	return s.Storer.Stat(ctx, key)
+}
+
+// TestShallowVerifyReportsStatErrorsApartFromMissing backs up a real
+// snapshot, deletes one blob, and makes storage fail Stat with an error
+// other than not-found for a second blob and for the encrypted database.
+// Only the deleted blob may be counted as missing. The second blob and
+// the database must be reported as could not be checked, with the error
+// storage returned, and verify must fail, in the text output and under
+// --json.
+func TestShallowVerifyReportsStatErrorsApartFromMissing(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 content, so the snapshot spans several blobs.
+	require.NoError(t, fs.MkdirAll(dataDir, 0o755))
+	appendRandomBytes(t, fs, map[string][]byte{},
+		filepath.Join(dataDir, "data.bin"), maxBlobSize*4)
+
+	ctx := context.Background()
+
+	cfg, storer, snapshotID := runFileStorageBackup(
+		ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
+
+	blobKeys := storedBlobKeys(t, storeDir)
+	require.GreaterOrEqual(t, len(blobKeys), 2, "expected at least two blobs")
+
+	missingKey, uncheckedKey := blobKeys[0], blobKeys[1]
+	missingHash, uncheckedHash := path.Base(missingKey), path.Base(uncheckedKey)
+
+	require.NoError(t, os.Remove(filepath.Join(storeDir, missingKey)))
+
+	dbKey := "metadata/" + snapshot.RemoteSnapshotKey(snapshotID) + "/db.zst.age"
+	failing := statFailingStorer{
+		Storer:   storer,
+		failKeys: []string{uncheckedKey, dbKey},
+	}
+
+	newVerifier := func(out io.Writer) *vaultik.Vaultik {
+		v := &vaultik.Vaultik{
+			Config:  cfg,
+			Storage: failing,
+			Fs:      fs,
+			Stdout:  out,
+			Stderr:  io.Discard,
+			UI:      ui.NewWithColor(io.Discard, false),
+		}
+		v.SetContext(ctx)
+
+		return v
+	}
+
+	var out bytes.Buffer
+
+	err := newVerifier(&out).VerifySnapshotWithOptions(
+		snapshotID, &vaultik.VerifyOptions{JSON: true})
+	require.Error(t, err)
+
+	result := decodeVerifyResult(t, out.Bytes())
+	require.Equal(t, "failed", result.Status)
+	require.Equal(t, 1, result.Missing, "only the deleted blob is missing")
+	require.Equal(t, result.BlobCount-2, result.Verified)
+	require.Len(t, result.Unchecked, 1)
+	require.Equal(t, uncheckedHash, result.Unchecked[0].Hash)
+	require.Contains(t, result.Unchecked[0].Error, errStatDenied.Error())
+	require.False(t, result.DatabaseMissing,
+		"a database that could not be checked is not missing")
+	require.Contains(t, result.DatabaseCheckError, errStatDenied.Error())
+
+	out.Reset()
+	err = newVerifier(&out).VerifySnapshotWithOptions(
+		snapshotID, &vaultik.VerifyOptions{})
+	require.Error(t, err)
+
+	text := out.String()
+	require.Contains(t, text, "Missing: "+missingHash)
+	require.NotContains(t, text, "Missing: "+uncheckedHash)
+	require.Contains(t, text, "Could not check: "+uncheckedHash)
+	require.Contains(t, text, "Encrypted database:       could not be checked")
+	require.Contains(t, text, errStatDenied.Error())
+	require.Contains(t, text, "FAILED")
+}
+
+// TestShallowVerifyQuietFailsOnStatErrorAlone backs up a healthy snapshot,
+// then makes storage fail Stat, with an error other than not-found, for
+// one blob alone and then for the encrypted database alone. Each run must
+// fail verify, and under --quiet, which drops the report, the error
+// storage returned must still reach stderr.
+//
+//nolint:paralleltest // captureProcessStderr replaces os.Stderr
+func TestShallowVerifyQuietFailsOnStatErrorAlone(t *testing.T) {
+	log.Initialize(log.Config{})
+
+	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)
+
+	require.NoError(t, fs.MkdirAll(dataDir, 0o755))
+	require.NoError(t, afero.WriteFile(fs,
+		filepath.Join(dataDir, "data.bin"),
+		bytesPattern("shallow-quiet-", int(maxBlobSize*2)), 0o644))
+
+	ctx := context.Background()
+
+	cfg, storer, snapshotID := runFileStorageBackup(
+		ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
+
+	newQuietVerifier := func(s storage.Storer) *vaultik.Vaultik {
+		v := &vaultik.Vaultik{
+			Config:  cfg,
+			Storage: s,
+			Fs:      fs,
+			Stdout:  io.Discard,
+			Stderr:  io.Discard,
+			UI:      ui.NewWithColor(io.Discard, false),
+		}
+		v.SetContext(ctx)
+		v.UI.SetQuiet(true)
+
+		return v
+	}
+
+	require.NoError(t,
+		newQuietVerifier(storer).VerifySnapshotWithOptions(
+			snapshotID, &vaultik.VerifyOptions{}),
+		"shallow verify should pass on a healthy snapshot")
+
+	blobKeys := storedBlobKeys(t, storeDir)
+	require.NotEmpty(t, blobKeys)
+
+	dbKey := "metadata/" + snapshot.RemoteSnapshotKey(snapshotID) + "/db.zst.age"
+
+	for _, failKey := range []string{blobKeys[0], dbKey} {
+		v := newQuietVerifier(statFailingStorer{
+			Storer:   storer,
+			failKeys: []string{failKey},
+		})
+
+		var err error
+
+		stderr := captureProcessStderr(t, func() {
+			err = v.VerifySnapshotWithOptions(
+				snapshotID, &vaultik.VerifyOptions{})
+		})
+
+		require.Error(t, err,
+			"verify must fail when %s could not be checked", failKey)
+		require.Contains(t, stderr, errStatDenied.Error(),
+			"under --quiet the error for %s must still reach stderr", failKey)
+	}
+}
+
+// storedBlobKeys returns the storage key of every blob file under
+// storeDir, sorted.
+func storedBlobKeys(t *testing.T, storeDir string) []string {
+	t.Helper()
+
+	var keys []string
+
+	err := filepath.Walk(filepath.Join(storeDir, "blobs"),
+		func(p string, info os.FileInfo, err error) error {
+			if err != nil {
+				return err
+			}
+
+			if info.IsDir() {
+				return nil
+			}
+
+			rel, err := filepath.Rel(storeDir, p)
+			if err != nil {
+				return err
+			}
+
+			keys = append(keys, filepath.ToSlash(rel))
+
+			return nil
+		})
+	require.NoError(t, err)
+
+	return keys
+}
+
 // growOneBlob appends bytes to the first blob file found under blobsDir,
 // changing its on-disk size so it no longer matches the manifest.
 func growOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {

internal/vaultik/snapshot.go

@@ -14,6 +14,7 @@ import (
 
 	"sneak.berlin/go/vaultik/internal/log"
 	"sneak.berlin/go/vaultik/internal/snapshot"
+	"sneak.berlin/go/vaultik/internal/storage"
 	"sneak.berlin/go/vaultik/internal/types"
 )
 
@@ -777,18 +778,10 @@ func (v *Vaultik) VerifySnapshotWithOptions(
 		v.stdoutf("Checking blob presence and sizes...\n")
 	}
 
-	// A snapshot is only restorable if its encrypted database is present
+	v.checkSnapshotDatabase(remoteKey, result)
-	// alongside the blobs. Shallow verify checks that the object exists; it
-	// does not decrypt it (that is deep verify's job).
-	dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey)
 
-	_, dbErr := v.Storage.Stat(v.ctx, dbPath)
+	result.Verified, result.Missing, result.Mismatched, result.MissingSize,
-	if dbErr != nil {
+		result.Unchecked, err = v.verifyManifestBlobs(manifest, opts)
-		result.DatabaseMissing = true
-	}
-
-	result.Verified, result.Missing, result.Mismatched, result.MissingSize, err =
-		v.verifyManifestBlobs(manifest, opts)
 	if err != nil {
 		if opts.JSON {
 			result.Status = verifyStatusFailed
@@ -825,25 +818,46 @@ func (v *Vaultik) printVerifyHeader(snapshotID string, opts *VerifyOptions) {
 	}
 }
 
+// checkSnapshotDatabase records in result whether the snapshot's
+// encrypted database is missing or could not be checked. A snapshot is
+// only restorable if its database is present alongside the blobs.
+// Shallow verify checks that the object exists; it does not decrypt it
+// (that is deep verify's job).
+func (v *Vaultik) checkSnapshotDatabase(remoteKey string, result *VerifyResult) {
+	dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey)
+
+	_, err := v.Storage.Stat(v.ctx, dbPath)
+
+	switch {
+	case errors.Is(err, storage.ErrNotFound):
+		result.DatabaseMissing = true
+	case err != nil:
+		result.DatabaseCheckError = err.Error()
+	}
+}
+
 // verifyManifestBlobs checks that each blob in the manifest is present in
 // storage with the size the manifest records, returning the counts of
 // blobs that were present with the right size, absent, and present but the
-// wrong size, plus the total bytes of the absent blobs. It does not read
+// wrong size, the total bytes of the absent blobs, and the blobs it could
-// blob contents; deep verification (RunDeepVerify) does that. The size
+// not check because storage returned an error other than not-found. It
-// comparison matches the deep path (see verifyBlobExistenceFromDB).
+// does not read blob contents; deep verification (RunDeepVerify) does
+// that. The size comparison matches the deep path (see
+// verifyBlobExistenceFromDB).
 func (v *Vaultik) verifyManifestBlobs(
 	manifest *snapshot.Manifest, opts *VerifyOptions,
-) (int, int, int, int64, error) {
+) (int, int, int, int64, []UncheckedBlob, error) {
 	var (
 		verified, missing, mismatched int
 		missingSize                   int64
+		unchecked                     []UncheckedBlob
 	)
 
 	for _, blob := range manifest.Blobs {
 		// The manifest is unauthenticated, so its blob hashes are checked
 		// before being spliced into a storage path.
 		if !isBlobHash(blob.Hash) {
-			return 0, 0, 0, 0, fmt.Errorf("%w: %s",
+			return 0, 0, 0, 0, nil, fmt.Errorf("%w: %s",
 				errInvalidBlobHash, shortHash(blob.Hash))
 		}
 
@@ -852,7 +866,7 @@ func (v *Vaultik) verifyManifestBlobs(
 
 		stat, err := v.Storage.Stat(v.ctx, blobPath)
 		switch {
-		case err != nil:
+		case errors.Is(err, storage.ErrNotFound):
 			if !opts.JSON && !v.UI.Quiet() {
 				v.stdoutf("  Missing: %s (%s)\n",
 					blob.Hash, ubytes(blob.CompressedSize))
@@ -860,6 +874,13 @@ func (v *Vaultik) verifyManifestBlobs(
 
 			missing++
 			missingSize += blob.CompressedSize
+		case err != nil:
+			if !opts.JSON && !v.UI.Quiet() {
+				v.stdoutf("  Could not check: %s (%v)\n", blob.Hash, err)
+			}
+
+			unchecked = append(unchecked,
+				UncheckedBlob{Hash: blob.Hash, Error: err.Error()})
 		case stat.Size != blob.CompressedSize:
 			if !opts.JSON && !v.UI.Quiet() {
 				v.stdoutf("  Wrong size: %s (store has %s, manifest lists %s)\n",
@@ -872,7 +893,7 @@ func (v *Vaultik) verifyManifestBlobs(
 		}
 	}
 
-	return verified, missing, mismatched, missingSize, nil
+	return verified, missing, mismatched, missingSize, unchecked, nil
 }
 
 // formatVerifyResult outputs the final verification results as JSON or
@@ -893,9 +914,19 @@ func (v *Vaultik) formatVerifyResult(
 		return v.outputVerifyJSON(result)
 	}
 
-	// Under --quiet a failure is still returned, and the cli layer
+	// --quiet drops the report but not the errors storage returned, which
-	// prints it on stderr.
+	// are logged on stderr. A failure is still returned either way, and
-	if !v.UI.Quiet() {
+	// the cli layer prints it on stderr.
+	if v.UI.Quiet() {
+		for _, blob := range result.Unchecked {
+			log.Warn("Could not check blob", "blob", blob.Hash, "error", blob.Error)
+		}
+
+		if result.DatabaseCheckError != "" {
+			log.Warn("Could not check the encrypted database",
+				"snapshot", result.SnapshotID, "error", result.DatabaseCheckError)
+		}
+	} else {
 		v.printVerifySummary(result, failure)
 	}
 
@@ -921,10 +952,19 @@ func (v *Vaultik) printVerifySummary(result *VerifyResult, failure string) {
 		v.stdoutf("  Wrong size:               %d blobs\n", result.Mismatched)
 	}
 
+	if len(result.Unchecked) > 0 {
+		v.stdoutf("  Could not check:          %d blobs\n", len(result.Unchecked))
+	}
+
 	if result.DatabaseMissing {
 		v.stdoutf("  Encrypted database:       missing\n")
 	}
 
+	if result.DatabaseCheckError != "" {
+		v.stdoutf("  Encrypted database:       could not be checked (%s)\n",
+			result.DatabaseCheckError)
+	}
+
 	v.stdoutf("  Status:       ")
 
 	if failure != "" {
@@ -952,10 +992,19 @@ func shallowVerifyFailure(result *VerifyResult) string {
 			fmt.Sprintf("%d blobs have the wrong size", result.Mismatched))
 	}
 
+	if len(result.Unchecked) > 0 {
+		parts = append(parts, fmt.Sprintf("%d blobs could not be checked",
+			len(result.Unchecked)))
+	}
+
 	if result.DatabaseMissing {
 		parts = append(parts, "the encrypted database is missing")
 	}
 
+	if result.DatabaseCheckError != "" {
+		parts = append(parts, "the encrypted database could not be checked")
+	}
+
 	return strings.Join(parts, "; ")
 }
 

internal/vaultik/symlinked_destination_test.go

@@ -0,0 +1,147 @@
+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")
+		})
+	}
+}
+
+// TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget checks that
+// snapshot remove warns that the metadata could not be removed from the
+// destination when metadata/ is a symlink whose target is missing, as it
+// does when the destination directory is missing, instead of finding no
+// metadata to remove.
+func TestRemoveSnapshotWarnsOnSymlinkedMetadataWithMissingTarget(t *testing.T) {
+	log.Initialize(log.Config{})
+	t.Parallel()
+
+	ctx := context.Background()
+	storeDir := filepath.Join(t.TempDir(), "store")
+	v, repos, out := backUpToFileDestination(ctx, t, storeDir)
+	targetPath := moveBehindSymlink(t, storeDir, "metadata")
+	require.NoError(t, os.Rename(targetPath, targetPath+"-unmounted"))
+
+	result, err := v.RemoveSnapshot(localSnapshotID(ctx, t, repos, "first"),
+		&vaultik.RemoveOptions{Force: true})
+	require.NoError(t, err)
+
+	assert.False(t, result.RemoteRemoved)
+	assert.Contains(t, out.String(),
+		"Could not remove snapshot metadata from remote")
+	assert.NotContains(t, out.String(),
+		"Removed snapshot metadata from remote storage")
+}
+
+// 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
+}

internal/vaultik/verify.go

@@ -57,11 +57,25 @@ type VerifyResult struct {
 	Missing     int    `json:"missing"`
 	MissingSize int64  `json:"missing_size,omitempty"`
 	Mismatched  int    `json:"mismatched,omitempty"`
+	// Unchecked lists the blobs shallow verify could not check, because
+	// storage answered with an error other than not-found, so whether
+	// they are present is unknown.
+	Unchecked []UncheckedBlob `json:"unchecked,omitempty"`
 	// DatabaseMissing is set by shallow verify when the snapshot's
 	// encrypted database (metadata/<key>/db.zst.age) is absent, which
 	// makes the snapshot unrestorable regardless of the blobs.
-	DatabaseMissing bool   `json:"database_missing,omitempty"`
+	DatabaseMissing bool `json:"database_missing,omitempty"`
-	ErrorMessage    string `json:"error,omitempty"`
+	// DatabaseCheckError is set by shallow verify to the error storage
+	// returned when the encrypted database could not be checked.
+	DatabaseCheckError string `json:"database_check_error,omitempty"`
+	ErrorMessage       string `json:"error,omitempty"`
+}
+
+// UncheckedBlob is a blob shallow verify could not check, with the error
+// storage returned for it.
+type UncheckedBlob struct {
+	Hash  string `json:"hash"`
+	Error string `json:"error"`
 }
 
 // deepVerifyFailure records a failure in the result and returns it appropriately