Trust only uploaded blobs for deduplication (closes #148)
check / check (pull_request) Successful in 2m52s

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.

Chosen 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. Option a (withholding all row writes until
the upload succeeds) would entangle the packer's writes with upload
ordering; gated reads plus a startup repair is smaller and self-healing.

Blobs recorded with no remote backend are marked uploaded so the
invariant holds uniformly.

model: claude-opus-4-8
This commit is contained in:
2026-09-21 23:22:29 +00:00
parent 343129f891
commit 1073420e8b
5 changed files with 314 additions and 7 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))
})