Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
07c957f84b |
@@ -412,7 +412,8 @@ destination store.
|
||||
entirely. Remote storage is unaffected; the next backup will do a full
|
||||
scan and re-deduplicate against existing remote blobs, and the local
|
||||
index will re-bind to the currently configured storage destination.
|
||||
Use this after changing `storage_url` to a different destination.
|
||||
Use this after changing `storage_url`, or `s3.endpoint`, `s3.bucket` or
|
||||
`s3.prefix`, to a different destination.
|
||||
* `--force`: Skip confirmation prompt
|
||||
|
||||
---
|
||||
|
||||
@@ -22,15 +22,14 @@ the tag exists and is exercised; what is left is merging `next` to
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-13: Recorded the size of the stored object for a blob a backup
|
||||
found already stored
|
||||
([issue #300](https://git.eeqj.de/sneak/vaultik/issues/300)). When the
|
||||
upload was skipped, the blob row and the manifest kept the size of the
|
||||
local copy, which differs from the stored object's when
|
||||
`compression_level` or the recipients changed since it was written.
|
||||
`snapshot verify` then failed on a good snapshot, shallow and deep,
|
||||
and the listed sizes were wrong. Marking the blob uploaded now records
|
||||
the size of the object in storage.
|
||||
- 2026-10-12: Bound the local index to the destination an `s3.*` config
|
||||
names ([issue #298](https://git.eeqj.de/sneak/vaultik/issues/298)).
|
||||
Only `storage_url` was checked, so after `s3.bucket` or `s3.prefix`
|
||||
changed, the next backup treated every chunk as already stored,
|
||||
uploaded nothing, and completed a snapshot that could not be restored.
|
||||
Without `storage_url`, the local index is now bound to `s3.endpoint`,
|
||||
`s3.bucket` and `s3.prefix`, and a change is refused as a changed
|
||||
`storage_url` is.
|
||||
|
||||
- 2026-10-12: Rejected a retention duration longer than about 292 years
|
||||
([issue #289](https://git.eeqj.de/sneak/vaultik/issues/289)). The
|
||||
|
||||
+2
-2
@@ -90,7 +90,7 @@ Stores information about packed, compressed, and encrypted blob files.
|
||||
- `created_ts` (INTEGER NOT NULL) - Creation timestamp
|
||||
- `finished_ts` (INTEGER) - Finalization timestamp (NULL if in progress)
|
||||
- `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. Once uploaded, the size of the object in storage: when the blob was already stored, under another compression level or other recipients, this differs from the size of the local copy
|
||||
- `compressed_size` (INTEGER NOT NULL DEFAULT 0) - Size after compression and encryption
|
||||
- `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.
|
||||
@@ -183,7 +183,7 @@ Tracks blob upload metrics.
|
||||
- `UPDATE blobs SET blob_hash = ?, finished_ts = ?` - Finalize blob after packing
|
||||
|
||||
4. **Upload**
|
||||
- `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)
|
||||
- `UPDATE blobs SET uploaded_ts = ?` - Mark blob as uploaded
|
||||
- `INSERT INTO uploads` - Record upload metrics with snapshot_id
|
||||
- `INSERT INTO snapshot_blobs` - Associate blob with snapshot
|
||||
|
||||
|
||||
@@ -159,16 +159,13 @@ func (r *BlobRepository) UpdateFinished(
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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.
|
||||
// UpdateUploaded marks a blob as uploaded
|
||||
func (r *BlobRepository) UpdateUploaded(
|
||||
ctx context.Context, tx *sql.Tx, id string, storedSize int64,
|
||||
ctx context.Context, tx *sql.Tx, id string,
|
||||
) error {
|
||||
query := `
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?, compressed_size = ?
|
||||
UPDATE blobs
|
||||
SET uploaded_ts = ?
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
@@ -176,9 +173,9 @@ func (r *BlobRepository) UpdateUploaded(
|
||||
|
||||
var err error
|
||||
if tx != nil {
|
||||
_, err = tx.ExecContext(ctx, query, now, storedSize, id)
|
||||
_, err = tx.ExecContext(ctx, query, now, id)
|
||||
} else {
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
|
||||
_, err = r.db.ExecWithLog(ctx, query, now, id)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
|
||||
@@ -122,8 +122,8 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
||||
t.Errorf("expected compressed size 500, got %d", updated.CompressedSize)
|
||||
}
|
||||
|
||||
// Test UpdateUploaded, with a stored object larger than the local copy
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600)
|
||||
// Test UpdateUploaded
|
||||
err = repo.UpdateUploaded(ctx, nil, blob.ID.String())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to update blob as uploaded: %v", err)
|
||||
}
|
||||
@@ -141,10 +141,6 @@ func TestBlobRepositoryUpdates(t *testing.T) {
|
||||
if uploaded.UploadedTS.Before(now.Add(-1 * time.Second)) {
|
||||
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) {
|
||||
|
||||
@@ -1509,8 +1509,7 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult)
|
||||
// snapshot references has uploaded_ts set, so deduplication and
|
||||
// interrupted-run repair treat these blobs as trustworthy.
|
||||
err = s.repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
|
||||
err := s.repos.Blobs.UpdateUploaded(
|
||||
ctx, tx, b.ID, b.Compressed)
|
||||
err := s.repos.Blobs.UpdateUploaded(ctx, tx, b.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marking blob uploaded: %w", err)
|
||||
}
|
||||
@@ -1546,7 +1545,7 @@ func (s *Scanner) handleBlobReady(
|
||||
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
|
||||
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
|
||||
|
||||
storedSize, blobExists, err := s.uploadBlobIfNeeded(
|
||||
blobExists, err := s.uploadBlobIfNeeded(
|
||||
ctx, blobPath, blobWithReader, startTime, result)
|
||||
if err != nil {
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
@@ -1554,8 +1553,7 @@ func (s *Scanner) handleBlobReady(
|
||||
return fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err)
|
||||
}
|
||||
|
||||
err = s.recordBlobMetadata(
|
||||
ctx, finishedBlob, storedSize, blobExists, startTime)
|
||||
err = s.recordBlobMetadata(ctx, finishedBlob, blobExists, startTime)
|
||||
if err != nil {
|
||||
s.cleanupBlobTempFile(blobWithReader)
|
||||
|
||||
@@ -1577,22 +1575,20 @@ func (s *Scanner) handleBlobReady(
|
||||
}
|
||||
|
||||
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
|
||||
// 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.
|
||||
// exist, returns whether it existed
|
||||
func (s *Scanner) uploadBlobIfNeeded(
|
||||
ctx context.Context,
|
||||
blobPath string,
|
||||
blobWithReader *blob.WithReader,
|
||||
startTime time.Time,
|
||||
result *ScanResult,
|
||||
) (int64, bool, error) {
|
||||
) (bool, error) {
|
||||
finishedBlob := blobWithReader.FinishedBlob
|
||||
|
||||
// Check if blob already exists (deduplication after restart)
|
||||
destination := s.storage.Info().Location
|
||||
|
||||
stored, err := s.storage.Stat(ctx, blobPath)
|
||||
_, err := s.storage.Stat(ctx, blobPath)
|
||||
if err == nil {
|
||||
log.Info("Blob already exists in storage, skipping upload",
|
||||
"hash", finishedBlob.Hash,
|
||||
@@ -1601,7 +1597,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
|
||||
s.ui.Path(destination))
|
||||
|
||||
return stored.Size, true, nil
|
||||
return true, nil
|
||||
}
|
||||
|
||||
s.ui.Beginf("Uploading blob %s (%s) to %s.",
|
||||
@@ -1615,7 +1611,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
if err != nil {
|
||||
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
|
||||
|
||||
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
return false, fmt.Errorf("uploading blob to storage: %w", err)
|
||||
}
|
||||
|
||||
uploadDuration := time.Since(startTime)
|
||||
@@ -1646,7 +1642,7 @@ func (s *Scanner) uploadBlobIfNeeded(
|
||||
stats.BytesUploaded.Add(finishedBlob.Compressed)
|
||||
}
|
||||
|
||||
return finishedBlob.Compressed, false, nil
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// makeUploadProgressCallback creates a progress callback for blob uploads.
|
||||
@@ -1716,7 +1712,6 @@ func (s *Scanner) makeUploadProgressCallback(
|
||||
func (s *Scanner) recordBlobMetadata(
|
||||
ctx context.Context,
|
||||
finishedBlob *blob.FinishedBlob,
|
||||
storedSize int64,
|
||||
blobExists bool,
|
||||
startTime time.Time,
|
||||
) error {
|
||||
@@ -1728,8 +1723,7 @@ func (s *Scanner) recordBlobMetadata(
|
||||
uploadDuration := time.Since(startTime)
|
||||
|
||||
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
|
||||
err := s.repos.Blobs.UpdateUploaded(
|
||||
txCtx, tx, finishedBlob.ID, storedSize)
|
||||
err := s.repos.Blobs.UpdateUploaded(txCtx, tx, finishedBlob.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("updating blob upload timestamp: %w", err)
|
||||
}
|
||||
|
||||
@@ -1,83 +0,0 @@
|
||||
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")
|
||||
}
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
// errStorageBindingMismatch is returned when the local index database is
|
||||
// bound to a different destination than the configured storage_url.
|
||||
// bound to a different destination than the configured one.
|
||||
var errStorageBindingMismatch = errors.New(
|
||||
"local index is bound to a different backup destination")
|
||||
|
||||
@@ -24,8 +24,9 @@ var errStorageBindingMismatch = errors.New(
|
||||
//
|
||||
// Behaviour:
|
||||
// - On first use (empty stored value), record the configured
|
||||
// storage_url and log the binding.
|
||||
// - When the stored value matches the configured storage_url, do
|
||||
// destination (storage_url, or the s3.* endpoint, bucket and
|
||||
// prefix) and log the binding.
|
||||
// - When the stored value matches the configured destination, do
|
||||
// nothing and return nil.
|
||||
// - When the two differ, refuse with an error that tells the user
|
||||
// how to recover (revert the config, or run `vaultik database
|
||||
@@ -46,12 +47,10 @@ func (v *Vaultik) EnsureStorageBinding() error {
|
||||
|
||||
configured := v.Config.StorageURL
|
||||
if configured == "" {
|
||||
// Some legacy configs still use the split s3.* keys instead of
|
||||
// storage_url. Falling back to a synthetic URL for those would
|
||||
// hide the fact that the binding is loose. Instead, treat
|
||||
// unset as "nothing to bind against" — the check is
|
||||
// necessarily best-effort for those configs.
|
||||
return nil
|
||||
// The s3.* form: bind on its endpoint, bucket and prefix,
|
||||
// written the way storage_url would name the same destination.
|
||||
s3 := v.Config.S3
|
||||
configured = "s3://" + s3.Bucket + "/" + s3.Prefix + "?endpoint=" + s3.Endpoint
|
||||
}
|
||||
|
||||
stored, err := v.Repositories.LocalMeta.Get(v.ctx, database.LocalMetaKeyStorageURL)
|
||||
@@ -96,7 +95,8 @@ func buildBindingMismatchMessage(stored, configured string) string {
|
||||
"future snapshots referencing blobs that don't exist at the new destination.\n" +
|
||||
"\n" +
|
||||
"To proceed, either:\n" +
|
||||
" - revert storage_url in your config to the bound destination, or\n" +
|
||||
" - revert storage_url (or the s3.* endpoint, bucket and prefix) in your\n" +
|
||||
" config to the bound destination, or\n" +
|
||||
" - run 'vaultik database delete' to discard the local index and rebuild it\n" +
|
||||
" from a fresh full backup against the new destination"
|
||||
}
|
||||
|
||||
@@ -76,3 +76,51 @@ func TestEnsureStorageBinding_MismatchRefuses(t *testing.T) {
|
||||
assert.Contains(t, err.Error(), "file:///mnt/backups/new")
|
||||
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||
}
|
||||
|
||||
// A config with no storage_url names its destination with s3.endpoint,
|
||||
// s3.bucket and s3.prefix; a change to any of them must be refused just
|
||||
// as a changed storage_url is.
|
||||
func TestEnsureStorageBinding_S3FieldsChangeRefuses(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
bound := config.S3Config{
|
||||
Endpoint: "https://s3.example.com",
|
||||
Bucket: "bucket-a",
|
||||
Prefix: "backups",
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
changed config.S3Config
|
||||
}{
|
||||
{"bucket", config.S3Config{
|
||||
Endpoint: bound.Endpoint, Bucket: "bucket-b", Prefix: bound.Prefix,
|
||||
}},
|
||||
{"prefix", config.S3Config{
|
||||
Endpoint: bound.Endpoint, Bucket: bound.Bucket, Prefix: "other",
|
||||
}},
|
||||
{"endpoint", config.S3Config{
|
||||
Endpoint: "https://s3.other.com", Bucket: bound.Bucket, Prefix: bound.Prefix,
|
||||
}},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tv, _ := buildBindTestVaultik(t, "")
|
||||
|
||||
tv.Config.S3 = bound
|
||||
require.NoError(t, tv.EnsureStorageBinding(), "first use binds")
|
||||
require.NoError(t, tv.EnsureStorageBinding(), "same destination passes")
|
||||
|
||||
tv.Config.S3 = tt.changed
|
||||
err := tv.EnsureStorageBinding()
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(),
|
||||
"local index is bound to a different backup destination")
|
||||
assert.Contains(t, err.Error(), "vaultik database delete")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user