Keep unreferenced blobs newer than prune_grace_period (closes #387)
check / check (push) Waiting to run

A backup uploads its blobs before the manifest that lists them, and the
process lock covers one user on one machine, so a prune from another host
or user sharing the destination deleted a running backup's blobs.

Prune now deletes an unreferenced blob only once its listed time is older
than the new config key prune_grace_period (default 24h; a negative value
is refused). A newer blob, or one listed with no time, is kept and counted
(blobs_too_new). A backup uploads a blob already on the destination again
instead of skipping it, so it gets a fresh time. The rclone backend takes
each object's time from the listing (use_server_modtime).

Cost: after database delete, the next backup uploads all its data again.
Side effect: a rebuilt blob replaces the stored copy; if compression_level
or the recipients changed since, older snapshots listing it fail verify.

Model: opus-5-5
This commit is contained in:
2026-10-14 08:57:35 +00:00
parent 2590f49583
commit 63c824220c
25 changed files with 639 additions and 114 deletions
+23 -42
View File
@@ -1641,8 +1641,7 @@ func (s *Scanner) handleBlobReady(
blobPath := fmt.Sprintf("blobs/%s/%s/%s",
finishedBlob.Hash[:2], finishedBlob.Hash[2:4], finishedBlob.Hash)
storedSize, blobExists, err := s.uploadBlobIfNeeded(
ctx, blobPath, blobWithReader, startTime, result)
err := s.uploadBlob(ctx, blobPath, blobWithReader, startTime, result)
if err != nil {
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
@@ -1650,8 +1649,7 @@ func (s *Scanner) handleBlobReady(
fmt.Errorf("uploading blob %s: %w", finishedBlob.Hash, err), cleanupErr)
}
err = s.recordBlobMetadata(
ctx, finishedBlob, storedSize, blobExists, startTime)
err = s.recordBlobMetadata(ctx, finishedBlob, startTime)
if err != nil {
cleanupErr := s.cleanupBlobTempFile(blobWithReader)
@@ -1675,46 +1673,33 @@ func (s *Scanner) handleBlobReady(
return nil
}
// uploadBlobIfNeeded uploads the blob to storage if it doesn't already
// exist. It returns the size of the object in storage, which for an
// existing object can differ from the local copy's, and whether it
// already existed.
func (s *Scanner) uploadBlobIfNeeded(
// uploadBlob uploads the blob to storage. A blob already there, left by
// an earlier run stopped before its manifest, is uploaded again rather
// than skipped: keeping its old time would let a prune elsewhere delete
// it as an unreferenced blob older than prune_grace_period before this
// backup's manifest lists it.
func (s *Scanner) uploadBlob(
ctx context.Context,
blobPath string,
blobWithReader *blob.WithReader,
startTime time.Time,
result *ScanResult,
) (int64, bool, error) {
) error {
finishedBlob := blobWithReader.FinishedBlob
// Check if blob already exists (deduplication after restart)
destination := s.storage.Info().Location
stored, err := s.storage.Stat(ctx, blobPath)
if err == nil {
log.Info("Blob already exists in storage, skipping upload",
"hash", finishedBlob.Hash,
"size", humanize.Bytes(safeUint64(finishedBlob.Compressed)))
s.ui.Infof("Blob %s (%s) already exists at %s. Skipping upload.",
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
s.ui.Path(destination))
return stored.Size, true, nil
}
s.ui.Beginf("Uploading blob %s (%s) to %s.",
s.ui.Hex(finishedBlob.Hash), s.ui.Size(finishedBlob.Compressed),
s.ui.Path(destination))
progressCallback := s.makeUploadProgressCallback(ctx, finishedBlob, startTime)
err = s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader,
err := s.storage.PutWithProgress(ctx, blobPath, blobWithReader.Reader,
finishedBlob.Compressed, progressCallback)
if err != nil {
log.Error("Failed to upload blob", "hash", finishedBlob.Hash, "error", err)
return 0, false, fmt.Errorf("uploading blob to storage: %w", err)
return fmt.Errorf("uploading blob to storage: %w", err)
}
uploadDuration := time.Since(startTime)
@@ -1745,7 +1730,7 @@ func (s *Scanner) uploadBlobIfNeeded(
stats.BytesUploaded.Add(finishedBlob.Compressed)
}
return finishedBlob.Compressed, false, nil
return nil
}
// makeUploadProgressCallback creates a progress callback for blob uploads.
@@ -1815,8 +1800,6 @@ func (s *Scanner) makeUploadProgressCallback(
func (s *Scanner) recordBlobMetadata(
ctx context.Context,
finishedBlob *blob.FinishedBlob,
storedSize int64,
blobExists bool,
startTime time.Time,
) error {
finishedBlobID, err := types.ParseBlobID(finishedBlob.ID)
@@ -1828,7 +1811,7 @@ func (s *Scanner) recordBlobMetadata(
return s.repos.WithTx(ctx, func(txCtx context.Context, tx *sql.Tx) error {
err := s.repos.Blobs.UpdateUploaded(
txCtx, tx, finishedBlob.ID, storedSize)
txCtx, tx, finishedBlob.ID, finishedBlob.Compressed)
if err != nil {
return fmt.Errorf("updating blob upload timestamp: %w", err)
}
@@ -1839,19 +1822,17 @@ func (s *Scanner) recordBlobMetadata(
return fmt.Errorf("adding blob to snapshot: %w", err)
}
if !blobExists {
upload := &database.Upload{
BlobHash: finishedBlob.Hash,
SnapshotID: s.snapshotID,
UploadedAt: startTime,
Size: finishedBlob.Compressed,
DurationMs: uploadDuration.Milliseconds(),
}
upload := &database.Upload{
BlobHash: finishedBlob.Hash,
SnapshotID: s.snapshotID,
UploadedAt: startTime,
Size: finishedBlob.Compressed,
DurationMs: uploadDuration.Milliseconds(),
}
err := s.repos.Uploads.Create(txCtx, tx, upload)
if err != nil {
return fmt.Errorf("recording upload metrics: %w", err)
}
err = s.repos.Uploads.Create(txCtx, tx, upload)
if err != nil {
return fmt.Errorf("recording upload metrics: %w", err)
}
return nil