Trust only uploaded blobs for deduplication (closes #148)
check / check (pull_request) Successful in 1m23s
check / check (push) Successful in 3m10s

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 the interrupted-upload test from #72: its t.Skip is removed and it passes against this fix, and this branch's earlier duplicate copy is dropped. The interrupted metadata-export case is split to #177.

Model: opus-4-8
This commit was merged in pull request #175.
This commit is contained in:
2026-09-22 10:29:21 +02:00
parent 6b7517a4dc
commit 38ebfd843a
5 changed files with 124 additions and 15 deletions
+57 -5
View File
@@ -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))
})