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

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.

Option b also names the interrupted metadata-export case (blobs
uploaded, manifest never written); that repair is split to a follow-up
issue.

Model: opus-4-8
This commit is contained in:
2026-09-22 07:35:01 +00:00
parent 994e5de613
commit 36d3791ed2
5 changed files with 314 additions and 7 deletions
+22 -2
View File
@@ -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)