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") +}