From 1073420e8ba2b3a6e03550f87c03700bfd0936c2 Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 21 Sep 2026 23:22:24 +0000 Subject: [PATCH] Trust only uploaded blobs for deduplication (closes #148) 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 --- TODO.md | 13 ++ internal/database/blobs.go | 24 +++ internal/database/chunks_ext.go | 24 ++- internal/snapshot/scanner.go | 62 +++++- internal/vaultik/interrupted_upload_test.go | 198 ++++++++++++++++++++ 5 files changed, 314 insertions(+), 7 deletions(-) create mode 100644 internal/vaultik/interrupted_upload_test.go diff --git a/TODO.md b/TODO.md index 28d4c28..5fd6eb7 100644 --- a/TODO.md +++ b/TODO.md @@ -25,6 +25,19 @@ release" is exactly the contradiction # 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 `prune` from reporting a failed row count as 0 ([issue #96](https://git.eeqj.de/sneak/vaultik/issues/96)). The seven `getTableCount` reads in `PruneDatabase` discarded their error, so a diff --git a/internal/database/blobs.go b/internal/database/blobs.go index 50c9862..dc85785 100644 --- a/internal/database/blobs.go +++ b/internal/database/blobs.go @@ -208,6 +208,30 @@ func (r *BlobRepository) DeleteOrphaned(ctx context.Context) error { 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 // (nil, nil) when no row matches. func (r *BlobRepository) getOne( diff --git a/internal/database/chunks_ext.go b/internal/database/chunks_ext.go index 4724acc..0fc9e0a 100644 --- a/internal/database/chunks_ext.go +++ b/internal/database/chunks_ext.go @@ -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) diff --git a/internal/snapshot/scanner.go b/internal/snapshot/scanner.go index 86dda30..bd85feb 100644 --- a/internal/snapshot/scanner.go +++ b/internal/snapshot/scanner.go @@ -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)) }) diff --git a/internal/vaultik/interrupted_upload_test.go b/internal/vaultik/interrupted_upload_test.go new file mode 100644 index 0000000..775740d --- /dev/null +++ b/internal/vaultik/interrupted_upload_test.go @@ -0,0 +1,198 @@ +package vaultik //nolint:testpackage // constructs Vaultik with unexported fields + +import ( + "bytes" + "context" + "errors" + "io" + "path/filepath" + "strings" + "testing" + + "github.com/spf13/afero" + "github.com/stretchr/testify/require" + "sneak.berlin/go/vaultik/internal/config" + "sneak.berlin/go/vaultik/internal/database" + "sneak.berlin/go/vaultik/internal/log" + "sneak.berlin/go/vaultik/internal/snapshot" + "sneak.berlin/go/vaultik/internal/storage" + "sneak.berlin/go/vaultik/internal/ui" +) + +// errUploadInterrupted stands in for a dropped connection or kill -9 +// partway through a blob upload. +var errUploadInterrupted = errors.New("simulated interrupted blob upload") + +// interruptedBlobStorer wraps a real Storer but fails every blob upload, +// modelling a run that dies mid-blob after the packer has already +// committed the blob's chunk rows to the local index. +type interruptedBlobStorer struct { + storage.Storer +} + +func (s *interruptedBlobStorer) PutWithProgress( + ctx context.Context, key string, reader io.Reader, + size int64, cb storage.ProgressCallback, +) error { + if strings.HasPrefix(key, "blobs/") { + return errUploadInterrupted + } + + return s.Storer.PutWithProgress(ctx, key, reader, size, cb) +} + +func (s *interruptedBlobStorer) Put( + ctx context.Context, key string, reader io.Reader, +) error { + if strings.HasPrefix(key, "blobs/") { + return errUploadInterrupted + } + + return s.Storer.Put(ctx, key, reader) +} + +// TestBackupRetryAfterInterruptedUploadIsRestorable reproduces the +// silent-data-loss defect in +// https://git.eeqj.de/sneak/vaultik/issues/148: a blob upload is +// interrupted, leaving chunk rows in the local index for data that never +// reached storage. The retry run reuses the same index. Before the fix it +// deduplicated against those orphaned chunks, uploaded nothing for them, +// and produced a snapshot that reported success but could not be +// restored. The retried snapshot must instead restore byte-for-byte. +func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) { + log.Initialize(log.Config{}) + t.Parallel() + + fs := afero.NewOsFs() + tempDir := t.TempDir() + + dataDir := filepath.Join(tempDir, "source") + storeDir := filepath.Join(tempDir, "remote") + restoreDir := filepath.Join(tempDir, "restored") + dbPath := filepath.Join(tempDir, "index.sqlite") + + require.NoError(t, fs.MkdirAll(dataDir, 0o755)) + + // Random content forces real chunks; small blobs guarantee at least + // one blob is finalized (and its upload attempted) during the run. + sources := map[string][]byte{ + "a.bin": randomBytes(t, 128*1024), + "b.bin": randomBytes(t, 128*1024), + "c.bin": randomBytes(t, 128*1024), + } + for name, data := range sources { + require.NoError(t, afero.WriteFile( + fs, filepath.Join(dataDir, name), data, 0o644)) + } + + cfg := &config.Config{ + AgeRecipients: []string{"age1ezrjmfpwsc95svdg0y54mums3zevgzu0x0ecq2" + + "f7tp8a05gl0sjq9q9wjg"}, + AgeSecretKey: "AGE-SECRET-KEY-19CR5YSFW59HM4TLD6GXVEDMZFTVVF7PPHKU" + + "T68TXSFPK7APHXA2QS2NJA5", + CompressionLevel: 3, + Hostname: "test-host", + ChunkSize: config.Size(16 * 1024), + BlobSizeLimit: config.Size(64 * 1024), + } + + working, err := storage.NewFileStorer(storeDir) + require.NoError(t, err) + + ctx := context.Background() + + // Run 1: the upload is interrupted, so the scan fails but leaves the + // interrupted blob's chunk rows committed in the index. + _, err = runBackup(ctx, fs, cfg, &interruptedBlobStorer{Storer: working}, + dataDir, dbPath, "interrupted") + require.Error(t, err, "interrupted upload must fail the run") + + // Run 2: retry on the same index with a working backend. This must + // succeed and produce a fully restorable snapshot. + snapshotID, err := runBackup(ctx, fs, cfg, working, dataDir, dbPath, "retry") + require.NoError(t, err, "retry backup must succeed") + + v := &Vaultik{ + Config: cfg, + Storage: working, + Fs: fs, + Stdout: io.Discard, + Stderr: io.Discard, + UI: ui.NewWithColor(io.Discard, false), + } + v.SetContext(ctx) + + require.NoError(t, v.Restore(&RestoreOptions{ + SnapshotID: snapshotID, + TargetDir: restoreDir, + }), "the retried snapshot must be restorable") + + for name, data := range sources { + restored := filepath.Join(restoreDir, dataDir, name) + got, err := afero.ReadFile(fs, restored) + require.NoErrorf(t, err, "restored file missing: %s", name) + require.Truef(t, bytes.Equal(got, data), + "restored bytes differ from original for %s", name) + } +} + +// runBackup performs one backup of dataDir into a fresh snapshot on the +// index at dbPath, returning the snapshot ID. When the scan succeeds it +// also completes and exports the snapshot metadata so the result can be +// restored. The index database is always closed before returning. +func runBackup( + ctx context.Context, + fs afero.Fs, + cfg *config.Config, + storer storage.Storer, + dataDir, dbPath, name string, +) (string, error) { + db, err := database.New(ctx, dbPath) + if err != nil { + return "", err + } + + defer func() { _ = db.Close() }() + + repos := database.NewRepositories(db) + + sm := snapshot.NewSnapshotManager(snapshot.SnapshotManagerParams{ + Repos: repos, + Storage: storer, + Config: cfg, + }) + sm.SetFilesystem(fs) + + scanner := snapshot.NewScanner(snapshot.ScannerConfig{ + FS: fs, + Storage: storer, + ChunkSize: cfg.ChunkSize.Int64(), + MaxBlobSize: cfg.BlobSizeLimit.Int64(), + CompressionLevel: cfg.CompressionLevel, + AgeRecipients: cfg.AgeRecipients, + Repositories: repos, + }) + + snapshotID, err := sm.CreateSnapshotWithName( + ctx, cfg.Hostname, name, "test-version", "test-git") + if err != nil { + return "", err + } + + _, err = scanner.Scan(ctx, dataDir, snapshotID) + if err != nil { + return snapshotID, err + } + + err = sm.CompleteSnapshot(ctx, snapshotID) + if err != nil { + return snapshotID, err + } + + err = sm.ExportSnapshotMetadata(ctx, dbPath, snapshotID) + if err != nil { + return snapshotID, err + } + + return snapshotID, nil +} -- 2.54.0