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

@@ -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`,

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(

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)

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

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