Trust only uploaded blobs for deduplication (closes #148) #175
@@ -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`,
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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))
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user