diff --git a/docs/DATAMODEL.md b/docs/DATAMODEL.md index b3191ac..463e6ec 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 can differ 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. @@ -184,7 +184,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 32441cf..a0bff48 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -1517,7 +1517,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) } @@ -1553,7 +1554,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) @@ -1561,7 +1562,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) @@ -1583,20 +1585,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, @@ -1605,7 +1609,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.", @@ -1619,7 +1623,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) @@ -1650,7 +1654,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. @@ -1720,6 +1724,7 @@ func (s *Scanner) makeUploadProgressCallback( func (s *Scanner) recordBlobMetadata( ctx context.Context, finishedBlob *blob.FinishedBlob, + storedSize int64, blobExists bool, startTime time.Time, ) error { @@ -1731,7 +1736,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") +}