Trust only uploaded blobs for deduplication (closes #148) #175

Merged
clawbot merged 1 commits from issue-148-interrupted-upload-dedup into next 2026-09-22 10:29:22 +02:00
5 changed files with 124 additions and 15 deletions
+13
View File
@@ -25,6 +25,19 @@ release" is exactly the contradiction
# Completed Steps # 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 - 2026-09-21: Stopped `--json` from silencing stderr diagnostics
([issue #112](https://git.eeqj.de/sneak/vaultik/issues/112)). `--json` ([issue #112](https://git.eeqj.de/sneak/vaultik/issues/112)). `--json`
used to be folded into `Quiet`, which pinned the log level to `WARN`, used to be folded into `Quiet`, which pinned the log level to `WARN`,
+24
View File
@@ -208,6 +208,30 @@ func (r *BlobRepository) DeleteOrphaned(ctx context.Context) error {
return nil 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 // getOne fetches a single blob row matched on the given column, or
// (nil, nil) when no row matches. // (nil, nil) when no row matches.
func (r *BlobRepository) getOne( func (r *BlobRepository) getOne(
+22 -2
View File
@@ -7,12 +7,32 @@ import (
// List returns every chunk in the index, ordered by chunk hash. // List returns every chunk in the index, ordered by chunk hash.
func (r *ChunkRepository) List(ctx context.Context) ([]*Chunk, error) { func (r *ChunkRepository) List(ctx context.Context) ([]*Chunk, error) {
query := ` return r.list(ctx, `
SELECT chunk_hash, size SELECT chunk_hash, size
FROM chunks FROM chunks
ORDER BY chunk_hash 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) rows, err := r.db.conn.QueryContext(ctx, query)
if err != nil { if err != nil {
return nil, fmt.Errorf("querying chunks: %w", err) return nil, fmt.Errorf("querying chunks: %w", err)
+57 -5
View File
@@ -220,7 +220,14 @@ func (s *Scanner) Scan(
defer s.progress.Stop() 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) knownFiles, err := s.loadDatabaseState(ctx, path)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -317,6 +324,38 @@ func (s *Scanner) loadDatabaseState(
return knownFiles, nil 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, // summarizeScanPhase calculates total size to process, updates progress tracking,
// and prints the scan phase summary with file counts and sizes // and prints the scan phase summary with file counts and sizes
func (s *Scanner) summarizeScanPhase( func (s *Scanner) summarizeScanPhase(
@@ -392,11 +431,14 @@ func (s *Scanner) loadKnownFiles(
return result, nil return result, nil
} }
// loadKnownChunks loads all known chunk hashes from the database into a // loadKnownChunks loads the chunk hashes safe to deduplicate against into
// map for fast lookup. This avoids per-chunk database queries during file // an in-memory map for fast lookup, avoiding per-chunk database queries
// processing. // 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 { 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 { if err != nil {
return fmt.Errorf("listing chunks: %w", err) 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) 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.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, return s.repos.Snapshots.AddBlob(ctx, tx, s.snapshotID, blobID,
types.BlobHash(b.Hash)) types.BlobHash(b.Hash))
}) })
+8 -8
View File
@@ -349,16 +349,16 @@ func TestInterruptedBlobUploadRecordsNoUploadedBlob(t *testing.T) {
} }
// Scenario 1b: after an interrupted upload, a retry on the same local // Scenario 1b: after an interrupted upload, a retry on the same local
// index must produce a restorable snapshot. It does not: the interrupted // index must produce a restorable snapshot. The interrupted run leaves
// run's chunk rows persist, the retry deduplicates against them, and the // the blob's chunk rows in the index; the fix for
// backup silently emits a snapshot referencing data never stored. Skipped // https://git.eeqj.de/sneak/vaultik/issues/148 discards those un-uploaded
// pending the fix. See https://git.eeqj.de/sneak/vaultik/issues/148. // 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 //nolint:paralleltest // installs the global logger via log.Initialize
func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) { 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{}) log.Initialize(log.Config{})
fs := afero.NewOsFs() fs := afero.NewOsFs()
@@ -418,7 +418,7 @@ func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) {
// with blobs and a database but no manifest. verify and snapshot list // with blobs and a database but no manifest. verify and snapshot list
// must report the damage honestly rather than crashing or passing. // must report the damage honestly rather than crashing or passing.
// Automatic detection and repair of this partial state on the next run // 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. // asserted here.
// //
//nolint:paralleltest // installs the global logger via log.Initialize //nolint:paralleltest // installs the global logger via log.Initialize