From e8ed3d0c27f2caa907713204833cd452aee4229f Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 13 Oct 2026 01:26:08 +0000 Subject: [PATCH] Record the stored size of a blob a backup finds already stored (closes #300) When a finished blob already existed at the destination, the upload was skipped but the blob row, and so the manifest, kept the size of the local copy. That copy differs from the stored object when compression_level or the recipients changed since it was written, so shallow and deep verify failed on a good snapshot. Marking a blob uploaded now records the size of the object in storage: for an existing blob, the size Stat reports. Judgement call: the test adds a recipient rather than changing the compression level, because an extra age recipient always makes the local copy larger, while two zstd levels need not compress the test data differently. Model: opus-5-5 --- TODO.md | 10 +++ docs/DATAMODEL.md | 4 +- internal/database/blobs.go | 15 ++-- internal/database/blobs_test.go | 8 +- internal/snapshot/scanner.go | 26 ++++--- internal/vaultik/reused_blob_verify_test.go | 83 +++++++++++++++++++++ 6 files changed, 126 insertions(+), 20 deletions(-) create mode 100644 internal/vaultik/reused_blob_verify_test.go diff --git a/TODO.md b/TODO.md index e124826..e898de8 100644 --- a/TODO.md +++ b/TODO.md @@ -22,6 +22,16 @@ 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-13: Made `snapshot remove` fail for a snapshot that does not exist ([issue #293](https://git.eeqj.de/sneak/vaultik/issues/293)). The command never checked the ID: it reported the snapshot removed diff --git a/docs/DATAMODEL.md b/docs/DATAMODEL.md index bcb9d1c..53a5056 100644 --- a/docs/DATAMODEL.md +++ b/docs/DATAMODEL.md @@ -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 +- `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 - `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 = ?` - 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 snapshot_blobs` - Associate blob with snapshot diff --git a/internal/database/blobs.go b/internal/database/blobs.go index dc85785..f9d56ba 100644 --- a/internal/database/blobs.go +++ b/internal/database/blobs.go @@ -159,13 +159,16 @@ func (r *BlobRepository) UpdateFinished( 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( - ctx context.Context, tx *sql.Tx, id string, + ctx context.Context, tx *sql.Tx, id string, storedSize int64, ) error { query := ` - UPDATE blobs - SET uploaded_ts = ? + UPDATE blobs + SET uploaded_ts = ?, compressed_size = ? WHERE id = ? ` @@ -173,9 +176,9 @@ func (r *BlobRepository) UpdateUploaded( var err error if tx != nil { - _, err = tx.ExecContext(ctx, query, now, id) + _, err = tx.ExecContext(ctx, query, now, storedSize, id) } else { - _, err = r.db.ExecWithLog(ctx, query, now, id) + _, err = r.db.ExecWithLog(ctx, query, now, storedSize, id) } if err != nil { diff --git a/internal/database/blobs_test.go b/internal/database/blobs_test.go index 457b0db..3caf57a 100644 --- a/internal/database/blobs_test.go +++ b/internal/database/blobs_test.go @@ -122,8 +122,8 @@ func TestBlobRepositoryUpdates(t *testing.T) { t.Errorf("expected compressed size 500, got %d", updated.CompressedSize) } - // Test UpdateUploaded - err = repo.UpdateUploaded(ctx, nil, blob.ID.String()) + // Test UpdateUploaded, with a stored object larger than the local copy + err = repo.UpdateUploaded(ctx, nil, blob.ID.String(), 600) if err != nil { 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)) { 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) { diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index c810278..672c52e 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -1509,7 +1509,8 @@ 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) + err := s.repos.Blobs.UpdateUploaded( + ctx, tx, b.ID, b.Compressed) if err != nil { return fmt.Errorf("marking blob uploaded: %w", err) } @@ -1545,7 +1546,7 @@ func (s *Scanner) handleBlobReady( blobPath := fmt.Sprintf("blobs/%s/%s/%s", finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash) - blobExists, err := s.uploadBlobIfNeeded( + storedSize, blobExists, err := s.uploadBlobIfNeeded( ctx, blobPath, blobWithReader, startTime, result) if err != nil { s.cleanupBlobTempFile(blobWithReader) @@ -1553,7 +1554,8 @@ func (s *Scanner) handleBlobReady( 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 { s.cleanupBlobTempFile(blobWithReader) @@ -1575,20 +1577,22 @@ func (s *Scanner) handleBlobReady( } // 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( ctx context.Context, blobPath string, blobWithReader *blob.WithReader, startTime time.Time, result *ScanResult, -) (bool, error) { +) (int64, bool, error) { finishedBlob := blobWithReader.FinishedBlob // Check if blob already exists (deduplication after restart) destination := s.storage.Info().Location - _, err := s.storage.Stat(ctx, blobPath) + stored, err := s.storage.Stat(ctx, blobPath) if err == nil { log.Info("Blob already exists in storage, skipping upload", "hash", finishedBlob.Hash, @@ -1597,7 +1601,7 @@ func (s *Scanner) uploadBlobIfNeeded( s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed), s.ui.Path(destination)) - return true, nil + return stored.Size, true, nil } s.ui.Beginf("Uploading blob %s (%s) to %s.", @@ -1611,7 +1615,7 @@ func (s *Scanner) uploadBlobIfNeeded( if err != nil { 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) @@ -1642,7 +1646,7 @@ func (s *Scanner) uploadBlobIfNeeded( stats.BytesUploaded.Add(finishedBlob.Compressed) } - return false, nil + return finishedBlob.Compressed, false, nil } // makeUploadProgressCallback creates a progress callback for blob uploads. @@ -1712,6 +1716,7 @@ func (s *Scanner) makeUploadProgressCallback( func (s *Scanner) recordBlobMetadata( ctx context.Context, finishedBlob *blob.FinishedBlob, + storedSize int64, blobExists bool, startTime time.Time, ) error { @@ -1723,7 +1728,8 @@ 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) + err := s.repos.Blobs.UpdateUploaded( + txCtx, tx, finishedBlob.ID, storedSize) if err != nil { return fmt.Errorf("updating blob upload timestamp: %w", err) } diff --git a/internal/vaultik/reused_blob_verify_test.go b/internal/vaultik/reused_blob_verify_test.go new file mode 100644 index 0000000..68cd715 --- /dev/null +++ b/internal/vaultik/reused_blob_verify_test.go @@ -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") +}