From 24f0dabf5103fdf56f0a0a08d957fb113fa3e7a4 Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 21 Sep 2026 23:22:24 +0000 Subject: [PATCH] Trust only uploaded blobs for deduplication (closes #148) An interrupted blob upload left the blob's chunks, blob_chunks, and blobs rows committed before the upload was attempted, so a later run deduplicated against data that never reached storage and produced a snapshot that reported success but could not be restored. Fix (issue option b): a chunk counts as known only when a blob holding it has uploaded_ts set, and each run drops un-uploaded blob rows and the chunks they orphan at startup, so the affected data is re-chunked and re-uploaded. A blob recorded with no remote backend is marked uploaded so the invariant holds uniformly. The reproduction is issue #72's interrupted-upload test: its t.Skip is removed and it passes against this fix, and this branch's earlier duplicate copy of it is dropped. The interrupted metadata-export case is split to follow-up issue #177. Model: opus-4-8 --- TODO.md | 13 +++++ internal/database/blobs.go | 24 +++++++++ internal/database/chunks_ext.go | 24 ++++++++- internal/snapshot/scanner.go | 62 ++++++++++++++++++++++-- internal/vaultik/fault_injection_test.go | 16 +++--- 5 files changed, 124 insertions(+), 15 deletions(-) diff --git a/TODO.md b/TODO.md index 0094121..0242f5a 100644 --- a/TODO.md +++ b/TODO.md @@ -25,6 +25,19 @@ release" is exactly the contradiction # Completed Steps +- 2026-09-21: Stopped an interrupted blob upload from making a later + backup deduplicate against data that was never stored + ([issue #148](https://git.eeqj.de/sneak/vaultik/issues/148)). The + packer commits a blob's `chunks`, `blob_chunks`, and `blobs` rows + before the upload is attempted, so a failed upload left chunk rows + behind and the next run skipped re-uploading them, producing a + snapshot that reported success but could not be restored. A run now + deduplicates only against chunks held by a blob whose `uploaded_ts` is + set, and at startup drops any un-uploaded blob rows (and the chunks + they orphan) so the affected data is re-chunked and re-uploaded. Blobs + recorded with no remote backend are marked uploaded so this invariant + holds uniformly. + - 2026-09-21: Stopped `--json` from silencing stderr diagnostics ([issue #112](https://git.eeqj.de/sneak/vaultik/issues/112)). `--json` used to be folded into `Quiet`, which pinned the log level to `WARN`, diff --git a/internal/database/blobs.go b/internal/database/blobs.go index 50c9862..dc85785 100644 --- a/internal/database/blobs.go +++ b/internal/database/blobs.go @@ -208,6 +208,30 @@ func (r *BlobRepository) DeleteOrphaned(ctx context.Context) error { return nil } +// DeleteUnuploaded deletes blob rows whose upload never completed +// (uploaded_ts IS NULL) and returns how many were removed. Their +// blob_chunks rows are removed by the ON DELETE CASCADE foreign key. +// A blob is only ever attached to a snapshot once its upload has been +// recorded, so an un-uploaded blob is never referenced by a completed +// snapshot: dropping it discards chunk rows that point at data which +// was never stored remotely, so the affected content is re-chunked and +// re-uploaded on the next run. +func (r *BlobRepository) DeleteUnuploaded(ctx context.Context) (int64, error) { + query := `DELETE FROM blobs WHERE uploaded_ts IS NULL` + + result, err := r.db.ExecWithLog(ctx, query) + if err != nil { + return 0, fmt.Errorf("deleting un-uploaded blobs: %w", err) + } + + rowsAffected, _ := result.RowsAffected() + if rowsAffected > 0 { + log.Debug("Deleted un-uploaded blobs", "count", rowsAffected) + } + + return rowsAffected, nil +} + // getOne fetches a single blob row matched on the given column, or // (nil, nil) when no row matches. func (r *BlobRepository) getOne( diff --git a/internal/database/chunks_ext.go b/internal/database/chunks_ext.go index 4724acc..0fc9e0a 100644 --- a/internal/database/chunks_ext.go +++ b/internal/database/chunks_ext.go @@ -7,12 +7,32 @@ import ( // List returns every chunk in the index, ordered by chunk hash. func (r *ChunkRepository) List(ctx context.Context) ([]*Chunk, error) { - query := ` + return r.list(ctx, ` SELECT chunk_hash, size FROM chunks ORDER BY chunk_hash - ` + `) +} +// ListInUploadedBlobs returns the chunks that are stored in a blob whose +// upload has completed (uploaded_ts set), ordered by chunk hash. These +// are the only chunks a backup may safely deduplicate against: a chunk +// recorded solely in a blob that was never uploaded refers to data that +// is not in remote storage, so trusting it would silently drop that data +// from later snapshots. +func (r *ChunkRepository) ListInUploadedBlobs(ctx context.Context) ([]*Chunk, error) { + return r.list(ctx, ` + SELECT DISTINCT c.chunk_hash, c.size + FROM chunks c + JOIN blob_chunks bc ON c.chunk_hash = bc.chunk_hash + JOIN blobs b ON bc.blob_id = b.id + WHERE b.uploaded_ts IS NOT NULL + ORDER BY c.chunk_hash + `) +} + +// list runs a chunk-selecting query and scans the (chunk_hash, size) rows. +func (r *ChunkRepository) list(ctx context.Context, query string) ([]*Chunk, error) { rows, err := r.db.conn.QueryContext(ctx, query) if err != nil { return nil, fmt.Errorf("querying chunks: %w", err) diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index 86dda30..bd85feb 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -220,7 +220,14 @@ func (s *Scanner) Scan( defer s.progress.Stop() } - // Phase 0: Load known files and chunks from database into memory for fast lookup + // Phase 0: Repair any state left by an interrupted previous run, then + // load known files and chunks from the database into memory for fast + // lookup. + err := s.repairInterruptedBlobs(ctx) + if err != nil { + return nil, err + } + knownFiles, err := s.loadDatabaseState(ctx, path) if err != nil { return nil, err @@ -317,6 +324,38 @@ func (s *Scanner) loadDatabaseState( return knownFiles, nil } +// repairInterruptedBlobs discards blob rows left by a previous run whose +// upload never completed. Such a blob has its chunks, blob_chunks, and +// blobs rows committed to the local index before the upload is attempted, +// so a crash or dropped connection mid-upload leaves them behind while the +// data never reaches remote storage. Deduplicating against those chunks on +// a later run would produce a snapshot that reports success but cannot be +// restored. Dropping the un-uploaded blobs (their blob_chunks cascade) and +// then any chunks left unreferenced forces the affected data to be +// re-chunked and re-uploaded this run. A blob is attached to a snapshot +// only once its upload is recorded, so this never touches a completed +// snapshot's data. +func (s *Scanner) repairInterruptedBlobs(ctx context.Context) error { + removed, err := s.repos.Blobs.DeleteUnuploaded(ctx) + if err != nil { + return fmt.Errorf("removing un-uploaded blob records: %w", err) + } + + if removed == 0 { + return nil + } + + log.Warn("Discarded blob records from an interrupted previous run; "+ + "their data will be re-uploaded", "blobs", removed) + + err = s.repos.Chunks.DeleteOrphaned(ctx) + if err != nil { + return fmt.Errorf("removing orphaned chunks: %w", err) + } + + return nil +} + // summarizeScanPhase calculates total size to process, updates progress tracking, // and prints the scan phase summary with file counts and sizes func (s *Scanner) summarizeScanPhase( @@ -392,11 +431,14 @@ func (s *Scanner) loadKnownFiles( return result, nil } -// loadKnownChunks loads all known chunk hashes from the database into a -// map for fast lookup. This avoids per-chunk database queries during file -// processing. +// loadKnownChunks loads the chunk hashes safe to deduplicate against into +// an in-memory map for fast lookup, avoiding per-chunk database queries +// during file processing. Only chunks held by a blob whose upload +// completed are loaded: a chunk left behind by an interrupted upload +// refers to data that never reached remote storage, and deduplicating +// against it would silently produce an unrestorable snapshot. func (s *Scanner) loadKnownChunks(ctx context.Context) error { - chunks, err := s.repos.Chunks.List(ctx) + chunks, err := s.repos.Chunks.ListInUploadedBlobs(ctx) if err != nil { return fmt.Errorf("listing chunks: %w", err) } @@ -1401,7 +1443,17 @@ func (s *Scanner) finalizeProcessPhase(ctx context.Context, result *ScanResult) return fmt.Errorf("parsing blob ID: %w", err) } + // With no remote backend the blob's lifecycle ends here, so + // mark it uploaded in the same transaction that attaches it to + // the snapshot. This keeps the invariant that any blob a + // 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) + if err != nil { + return fmt.Errorf("marking blob uploaded: %w", err) + } + return s.repos.Snapshots.AddBlob(ctx, tx, s.snapshotID, blobID, types.BlobHash(b.Hash)) }) diff --git a/internal/vaultik/fault_injection_test.go b/internal/vaultik/fault_injection_test.go index 88d30cf..9e39096 100644 --- a/internal/vaultik/fault_injection_test.go +++ b/internal/vaultik/fault_injection_test.go @@ -349,16 +349,16 @@ func TestInterruptedBlobUploadRecordsNoUploadedBlob(t *testing.T) { } // Scenario 1b: after an interrupted upload, a retry on the same local -// index must produce a restorable snapshot. It does not: the interrupted -// run's chunk rows persist, the retry deduplicates against them, and the -// backup silently emits a snapshot referencing data never stored. Skipped -// pending the fix. See https://git.eeqj.de/sneak/vaultik/issues/148. +// index must produce a restorable snapshot. The interrupted run leaves +// the blob's chunk rows in the index; the fix for +// https://git.eeqj.de/sneak/vaultik/issues/148 discards those un-uploaded +// blob rows at the start of the next scan and deduplicates only against +// chunks in a blob that was actually uploaded, so the retry re-chunks and +// re-uploads the affected data instead of silently referencing data that +// never reached storage. // //nolint:paralleltest // installs the global logger via log.Initialize func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) { - t.Skip("blocked on https://git.eeqj.de/sneak/vaultik/issues/148: " + - "retry after an interrupted upload silently produces an " + - "unrestorable snapshot") log.Initialize(log.Config{}) fs := afero.NewOsFs() @@ -418,7 +418,7 @@ func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) { // with blobs and a database but no manifest. verify and snapshot list // must report the damage honestly rather than crashing or passing. // Automatic detection and repair of this partial state on the next run -// is tracked in https://git.eeqj.de/sneak/vaultik/issues/148 and is not +// is tracked in https://git.eeqj.de/sneak/vaultik/issues/177 and is not // asserted here. // //nolint:paralleltest // installs the global logger via log.Initialize