Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
96b134e5cc |
@@ -22,17 +22,6 @@ the tag exists and is exercised; what is left is merging `next` to
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
- 2026-10-13: Made a `file://` destination list through a symlinked
|
|
||||||
directory ([issue #296](https://git.eeqj.de/sneak/vaultik/issues/296)).
|
|
||||||
A `metadata/` or `blobs/` directory that was a symlink to a directory
|
|
||||||
elsewhere was listed as one file, so the destination store listed as
|
|
||||||
empty while backups and restores still worked through the link.
|
|
||||||
`prune` then removed the local snapshot records and deleted every blob,
|
|
||||||
and `remote info` and `remote nuke` found no blobs. The listings now
|
|
||||||
read through the link as `Get` and `Stat` do. A link whose target is
|
|
||||||
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
|
- 2026-10-13: Stopped `remote info` counting a file under `blobs/` that
|
||||||
is not a blob as an orphaned blob
|
is not a blob as an orphaned blob
|
||||||
([issue #295](https://git.eeqj.de/sneak/vaultik/issues/295)). A stray
|
([issue #295](https://git.eeqj.de/sneak/vaultik/issues/295)). A stray
|
||||||
|
|||||||
+2
-2
@@ -90,7 +90,7 @@ Stores information about packed, compressed, and encrypted blob files.
|
|||||||
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
||||||
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
||||||
- `uncompressed_size` (INTEGER NOT NULL DEFAULT 0) - Total size of chunks before compression
|
- `uncompressed_size` (INTEGER NOT NULL DEFAULT 0) - Total size of chunks before compression
|
||||||
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption
|
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption. Once uploaded, the size of the object in storage: when the blob was already stored, under another compression level or other recipients, this can differ from the size of the local copy
|
||||||
- `uploaded_ts` (INTEGER) - Upload completion timestamp (NULL if not uploaded)
|
- `uploaded_ts` (INTEGER) - Upload completion timestamp (NULL if not uploaded)
|
||||||
|
|
||||||
**Purpose:** Tracks blob lifecycle from creation through upload. The UUID primary key allows immediate association of chunks with blobs.
|
**Purpose:** Tracks blob lifecycle from creation through upload. The UUID primary key allows immediate association of chunks with blobs.
|
||||||
@@ -184,7 +184,7 @@ Tracks blob upload metrics.
|
|||||||
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
||||||
|
|
||||||
4. **Upload**
|
4. **Upload**
|
||||||
- `UPDATE blobs SET uploaded_ts = ?` - Mark blob as uploaded
|
- `UPDATE blobs SET uploaded_ts = ?, compressed_size = ?` - Mark blob as uploaded, with the size of the object in storage (the upload is skipped when the blob is already stored)
|
||||||
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
||||||
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
||||||
|
|
||||||
|
|||||||
@@ -159,13 +159,16 @@ func (r *BlobRepository) UpdateFinished(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateUploaded marks a blob as uploaded
|
// UpdateUploaded marks a blob as uploaded and records the size of the
|
||||||
|
// stored object as its compressed size. That size can differ from the
|
||||||
|
// local copy's when an identical blob was already stored with another
|
||||||
|
// compression level or other recipients.
|
||||||
func (r *BlobRepository) UpdateUploaded(
|
func (r *BlobRepository) UpdateUploaded(
|
||||||
ctx context.Context, tx *sql.Tx, id string,
|
ctx context.Context, tx *sql.Tx, id string, storedSize int64,
|
||||||
) error {
|
) error {
|
||||||
query := `
|
query := `
|
||||||
UPDATE blobs
|
UPDATE blobs
|
||||||
SET uploaded_ts = ?
|
SET uploaded_ts = ?, compressed_size = ?
|
||||||
WHERE id = ?
|
WHERE id = ?
|
||||||
`
|
`
|
||||||
|
|
||||||
@@ -173,9 +176,9 @@ func (r *BlobRepository) UpdateUploaded(
|
|||||||
|
|
||||||
var err error
|
var err error
|
||||||
if tx != nil {
|
if tx != nil {
|
||||||
_, err = tx.ExecContext(ctx, query, now, id)
|
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
||||||
} else {
|
} else {
|
||||||
_, err = r.db.ExecWithLog(ctx, query, now, id)
|
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -122,8 +122,8 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
|||||||
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Test UpdateUploaded
|
// Test UpdateUploaded, with a stored object larger than the local copy
|
||||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("failed to update blob as uploaded: %v", err)
|
t.Fatalf("failed to update blob as uploaded: %v", err)
|
||||||
}
|
}
|
||||||
@@ -141,6 +141,10 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
|||||||
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
||||||
t.Error("uploaded timestamp should be around test time")
|
t.Error("uploaded timestamp should be around test time")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if uploaded.CompressedSize != 600 {
|
||||||
|
t.Errorf("expected the stored size 600, got %d", uploaded.CompressedSize)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestBlobRepositoryDuplicate(t *testing.T) {
|
func TestBlobRepositoryDuplicate(t *testing.T) {
|
||||||
|
|||||||
@@ -1517,7 +1517,8 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
|
|||||||
// snapshot references has uploaded_ts set, so deduplication and
|
// snapshot references has uploaded_ts set, so deduplication and
|
||||||
// interrupted-run repair treat these blobs as trustworthy.
|
// interrupted-run repair treat these blobs as trustworthy.
|
||||||
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||||
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
err := s.repos.Blobs.UpdateUploaded(
|
||||||
|
ctx, tx, b.ID, b.Compressed)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||||
}
|
}
|
||||||
@@ -1553,7 +1554,7 @@ func (s *Scanner) handleBlobReady(
|
|||||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
||||||
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
||||||
|
|
||||||
blobExists, err := s.uploadBlobIfNeeded(
|
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
||||||
ctx, blobPath, blobWithReader, startTime, result)
|
ctx, blobPath, blobWithReader, startTime, result)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.cleanupBlobTempFile(blobWithReader)
|
s.cleanupBlobTempFile(blobWithReader)
|
||||||
@@ -1561,7 +1562,8 @@ func (s *Scanner) handleBlobReady(
|
|||||||
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
err = s.recordBlobMetadata(
|
||||||
|
ctx, finishedBlob, storedSize, blobExists, startTime)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.cleanupBlobTempFile(blobWithReader)
|
s.cleanupBlobTempFile(blobWithReader)
|
||||||
|
|
||||||
@@ -1583,20 +1585,22 @@ func (s *Scanner) handleBlobReady(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
||||||
// exist, returns whether it existed
|
// exist. It returns the size of the object in storage, which for an
|
||||||
|
// existing object can differ from the local copy's, and whether it
|
||||||
|
// already existed.
|
||||||
func (s *Scanner) uploadBlobIfNeeded(
|
func (s *Scanner) uploadBlobIfNeeded(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
blobPath string,
|
blobPath string,
|
||||||
blobWithReader *blob.WithReader,
|
blobWithReader *blob.WithReader,
|
||||||
startTime time.Time,
|
startTime time.Time,
|
||||||
result *ScanResult,
|
result *ScanResult,
|
||||||
) (bool, error) {
|
) (int64, bool, error) {
|
||||||
finishedBlob := blobWithReader.FinishedBlob
|
finishedBlob := blobWithReader.FinishedBlob
|
||||||
|
|
||||||
// Check if blob already exists (deduplication after restart)
|
// Check if blob already exists (deduplication after restart)
|
||||||
destination := s.storage.Info().Location
|
destination := s.storage.Info().Location
|
||||||
|
|
||||||
_, err := s.storage.Stat(ctx, blobPath)
|
stored, err := s.storage.Stat(ctx, blobPath)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
log.Info("Blob already exists in storage, skipping upload",
|
log.Info("Blob already exists in storage, skipping upload",
|
||||||
"hash", finishedBlob.Hash,
|
"hash", finishedBlob.Hash,
|
||||||
@@ -1605,7 +1609,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
||||||
s.ui.Path(destination))
|
s.ui.Path(destination))
|
||||||
|
|
||||||
return true, nil
|
return stored.Size, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||||
@@ -1619,7 +1623,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
||||||
|
|
||||||
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
uploadDuration := time.Since(startTime)
|
uploadDuration := time.Since(startTime)
|
||||||
@@ -1650,7 +1654,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
|||||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||||
}
|
}
|
||||||
|
|
||||||
return false, nil
|
return finishedBlob.Compressed, false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||||
@@ -1720,6 +1724,7 @@ func (s *Scanner) makeUploadProgressCallback(
|
|||||||
func (s *Scanner) recordBlobMetadata(
|
func (s *Scanner) recordBlobMetadata(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
finishedBlob *blob.FinishedBlob,
|
finishedBlob *blob.FinishedBlob,
|
||||||
|
storedSize int64,
|
||||||
blobExists bool,
|
blobExists bool,
|
||||||
startTime time.Time,
|
startTime time.Time,
|
||||||
) error {
|
) error {
|
||||||
@@ -1731,7 +1736,8 @@ func (s *Scanner) recordBlobMetadata(
|
|||||||
uploadDuration := time.Since(startTime)
|
uploadDuration := time.Since(startTime)
|
||||||
|
|
||||||
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
||||||
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
|
err := s.repos.Blobs.UpdateUploaded(
|
||||||
|
txCtx, tx, finishedBlob.ID, storedSize)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
+10
-45
@@ -120,10 +120,9 @@ func (f *FileStorer) Delete(_ context.Context, key string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// List returns all keys with the given prefix, reading through symlinked
|
// List returns all keys with the given prefix. It fails when the
|
||||||
// directories. It fails when the destination directory is missing or the
|
// destination directory is missing; a missing prefix under it is an
|
||||||
// prefix is a symlink whose target is missing; a missing prefix under the
|
// empty listing.
|
||||||
// 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
|
||||||
|
|
||||||
@@ -134,7 +133,8 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
exists, err := f.prefixExists(basePath)
|
// Check if the prefix exists
|
||||||
|
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 = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
err = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -176,9 +176,7 @@ 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 reads through symlinked directories and sends an error when
|
// List, it sends an error when the destination directory is missing.
|
||||||
// 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)
|
||||||
|
|
||||||
@@ -194,7 +192,8 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
|
|||||||
|
|
||||||
basePath := f.fullPath(prefix)
|
basePath := f.fullPath(prefix)
|
||||||
|
|
||||||
exists, err := f.prefixExists(basePath)
|
// Check if the prefix exists
|
||||||
|
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)}
|
||||||
|
|
||||||
@@ -205,7 +204,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
|
||||||
}
|
}
|
||||||
|
|
||||||
_ = f.walk(basePath, func(path string, info os.FileInfo, err error) error {
|
_ = afero.Walk(f.fs, basePath, func(path string, info os.FileInfo, err error) error {
|
||||||
// Check context cancellation
|
// Check context cancellation
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
@@ -353,40 +352,6 @@ 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,10 +4,7 @@ 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"
|
||||||
@@ -63,151 +60,3 @@ 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")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -0,0 +1,83 @@
|
|||||||
|
package vaultik_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"filippo.io/age"
|
||||||
|
"github.com/spf13/afero"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"sneak.berlin/go/vaultik/internal/log"
|
||||||
|
"sneak.berlin/go/vaultik/internal/ui"
|
||||||
|
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestVerifyPassesWhenBackupReusesStoredBlob backs up a file, then backs it
|
||||||
|
// up again from an empty local index, as after `database delete`, with an
|
||||||
|
// extra age recipient. The second backup builds the same blob, finds it
|
||||||
|
// already stored and skips the upload, but its own copy is larger, because
|
||||||
|
// age writes a header entry for each recipient. The second snapshot must
|
||||||
|
// record the size of the stored object, so shallow and deep verify pass.
|
||||||
|
func TestVerifyPassesWhenBackupReusesStoredBlob(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")
|
||||||
|
secondDBPath := filepath.Join(tempDir, "second.sqlite")
|
||||||
|
|
||||||
|
chunkSize := int64(64 * 1024)
|
||||||
|
maxBlobSize := int64(512 * 1024)
|
||||||
|
|
||||||
|
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
|
||||||
|
require.NoError(t, afero.WriteFile(fs,
|
||||||
|
filepath.Join(dataDir, "data.bin"),
|
||||||
|
bytesPattern("reuse-", int(chunkSize*3)), 0o644))
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
runFileStorageBackup(ctx, t, fs, dataDir, storeDir,
|
||||||
|
filepath.Join(tempDir, "first.sqlite"), chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
env := setupDedupBackupEnv(
|
||||||
|
ctx, t, fs, storeDir, secondDBPath, chunkSize, maxBlobSize)
|
||||||
|
|
||||||
|
defer func() { _ = env.db.Close() }()
|
||||||
|
|
||||||
|
extra, err := age.GenerateX25519Identity()
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
env.cfg.AgeRecipients = append(env.cfg.AgeRecipients,
|
||||||
|
extra.Recipient().String())
|
||||||
|
|
||||||
|
snapshotID, result := runDedupSnapshot(ctx, t, env.sm, env.makeScanner(),
|
||||||
|
env.cfg.Hostname, "second", dataDir, secondDBPath)
|
||||||
|
require.Positive(t, result.BlobsCreated)
|
||||||
|
require.Zero(t, result.BlobsUploaded,
|
||||||
|
"the second backup should find every blob already stored")
|
||||||
|
|
||||||
|
require.NoError(t, env.db.Close())
|
||||||
|
|
||||||
|
verifier := &vaultik.Vaultik{
|
||||||
|
Config: env.cfg,
|
||||||
|
Storage: env.storer,
|
||||||
|
Fs: fs,
|
||||||
|
Stdout: io.Discard,
|
||||||
|
Stderr: io.Discard,
|
||||||
|
UI: ui.NewWithColor(io.Discard, false),
|
||||||
|
}
|
||||||
|
verifier.SetContext(ctx)
|
||||||
|
|
||||||
|
require.NoError(t,
|
||||||
|
verifier.VerifySnapshotWithOptions(snapshotID, &vaultik.VerifyOptions{}),
|
||||||
|
"shallow verify should pass")
|
||||||
|
require.NoError(t,
|
||||||
|
verifier.VerifySnapshotWithOptions(snapshotID,
|
||||||
|
&vaultik.VerifyOptions{Deep: true}),
|
||||||
|
"deep verify should pass")
|
||||||
|
}
|
||||||
@@ -1,121 +0,0 @@
|
|||||||
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