Keep unreferenced blobs newer than prune_grace_period (closes #387)

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). The rclone backend takes each object's time from the
listing (use_server_modtime).

Decision: a blob a backup reuses from an earlier run that stopped before
its manifest keeps its old time and is not protected; the docs and config
comments say so.

Model: opus-5-5

internal/vaultik/info.go

@@ -457,9 +457,9 @@ func (v *Vaultik) scanRemoteBlobStorage(
 		return err
 	}
 
-	for _, size := range allBlobs {
+	for _, blob := range allBlobs {
 		result.TotalBlobCount++
-		result.TotalBlobSize += size
+		result.TotalBlobSize += blob.Size
 	}
 
 	// A blob named only by a manifest that could not be read, or by one
@@ -474,10 +474,10 @@ func (v *Vaultik) scanRemoteBlobStorage(
 		orphanedSize  int64
 	)
 
-	for hash, size := range allBlobs {
+	for hash, blob := range allBlobs {
 		if _, referenced := referencedBlobs[hash]; !referenced {
 			orphanedCount++
-			orphanedSize += size
+			orphanedSize += blob.Size
 		}
 	}
 

internal/vaultik/prune.go

@@ -5,8 +5,10 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"sneak.berlin/go/vaultik/internal/log"
+	"sneak.berlin/go/vaultik/internal/storage"
 )
 
 // PruneOptions contains options for the prune command
@@ -163,6 +165,10 @@ type PruneBlobsResult struct {
 	BlobsDeleted int   `json:"blobs_deleted"`
 	BlobsFailed  int   `json:"blobs_failed,omitempty"`
 	BytesFreed   int64 `json:"bytes_freed"`
+	// BlobsTooNew counts the unreferenced blobs kept because they are
+	// newer than the prune grace period, or of unknown age. BlobsFound
+	// does not include them.
+	BlobsTooNew int `json:"blobs_too_new"`
 }
 
 // Prune removes orphaned data from the local index database AND
@@ -208,9 +214,15 @@ func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
 		return err
 	}
 
-	unreferencedBlobs, totalSize := v.findUnreferencedBlobs(allBlobs, allBlobsReferenced)
+	unreferencedBlobs, totalSize, tooNew := v.findUnreferencedBlobs(
+		allBlobs, allBlobsReferenced)
 
-	result := &PruneBlobsResult{BlobsFound: len(unreferencedBlobs)}
+	result := &PruneBlobsResult{
+		BlobsFound:  len(unreferencedBlobs),
+		BlobsTooNew: tooNew,
+	}
+
+	v.reportBlobsTooNew(tooNew, opts)
 
 	if len(unreferencedBlobs) == 0 {
 		log.Info("No unreferenced blobs found")
@@ -344,11 +356,12 @@ func (v *Vaultik) listUniqueSnapshotIDs() ([]string, error) {
 	return snapshotIDs, nil
 }
 
-// listAllRemoteBlobs returns a map of all blob hashes to their sizes in remote storage
-func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) {
+// listAllRemoteBlobs returns a map of all blob hashes in remote storage
+// to their listing entries.
+func (v *Vaultik) listAllRemoteBlobs() (map[string]storage.ObjectInfo, error) {
 	log.Info("Listing all blobs in storage")
 
-	allBlobs := make(map[string]int64)
+	allBlobs := make(map[string]storage.ObjectInfo)
 	blobObjectCh := v.Storage.ListStream(v.ctx, "blobs/")
 
 	for object := range blobObjectCh {
@@ -372,7 +385,7 @@ func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) {
 			continue
 		}
 
-		allBlobs[parts[3]] = object.Size
+		allBlobs[parts[3]] = object
 	}
 
 	log.Info("Found blobs in storage", "count", len(allBlobs))
@@ -380,30 +393,59 @@ func (v *Vaultik) listAllRemoteBlobs() (map[string]int64, error) {
 	return allBlobs, nil
 }
 
-// findUnreferencedBlobs returns blob hashes not referenced by any
-// manifest and their total size.
+// findUnreferencedBlobs returns the hashes of the blobs no manifest
+// references that are older than the prune grace period, and their total
+// size. It also returns how many unreferenced blobs it left out for
+// being newer than that, or of unknown age because the listing gave no
+// time for them.
 func (v *Vaultik) findUnreferencedBlobs(
-	allBlobs map[string]int64, referenced map[string]bool,
-) ([]string, int64) {
+	allBlobs map[string]storage.ObjectInfo, referenced map[string]bool,
+) ([]string, int64, int) {
 	var (
 		unreferenced []string
 		totalSize    int64
+		tooNew       int
 	)
 
-	for hash, size := range allBlobs {
-		if !referenced[hash] {
-			unreferenced = append(unreferenced, hash)
-			totalSize += size
+	cutoff := time.Now().UTC().Add(-v.Config.PruneGracePeriod)
+
+	for hash, blob := range allBlobs {
+		if referenced[hash] {
+			continue
 		}
+
+		if blob.ModTime.IsZero() || blob.ModTime.After(cutoff) {
+			tooNew++
+
+			continue
+		}
+
+		unreferenced = append(unreferenced, hash)
+		totalSize += blob.Size
 	}
 
-	return unreferenced, totalSize
+	return unreferenced, totalSize, tooNew
+}
+
+// reportBlobsTooNew says how many unreferenced blobs prune kept for being
+// newer than the grace period or of unknown age. Under --json the
+// document carries the count instead.
+func (v *Vaultik) reportBlobsTooNew(tooNew int, opts *PruneOptions) {
+	if tooNew == 0 || opts.JSON {
+		return
+	}
+
+	v.UI.Infof("Kept %d unreferenced blob(s) newer than "+
+		"prune_grace_period (%s), or of unknown age: a backup still "+
+		"running may need them.",
+		tooNew, v.Config.PruneGracePeriod)
 }
 
 // deleteUnreferencedBlobs deletes the given blobs from storage and
 // populates the result.
 func (v *Vaultik) deleteUnreferencedBlobs(
-	unreferencedBlobs []string, allBlobs map[string]int64, result *PruneBlobsResult,
+	unreferencedBlobs []string, allBlobs map[string]storage.ObjectInfo,
+	result *PruneBlobsResult,
 ) {
 	log.Info("Deleting unreferenced blobs")
 
@@ -418,7 +460,7 @@ func (v *Vaultik) deleteUnreferencedBlobs(
 		}
 
 		result.BlobsDeleted++
-		result.BytesFreed += allBlobs[hash]
+		result.BytesFreed += allBlobs[hash].Size
 
 		if (i+1)%progressLogEvery == 0 || i == len(unreferencedBlobs)-1 {
 			log.Info("Deletion progress",

internal/vaultik/prune_grace_period_test.go

@@ -0,0 +1,166 @@
+package vaultik_test
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"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/storage"
+	"sneak.berlin/go/vaultik/internal/storage/faultstore"
+	"sneak.berlin/go/vaultik/internal/ui"
+	"sneak.berlin/go/vaultik/internal/vaultik"
+)
+
+// TestPruneDuringBackupKeepsItsBlobs runs a prune from another host
+// sharing the destination store at the moment a backup has uploaded its
+// blobs but not yet its manifest, so no manifest lists those blobs. The
+// prune must leave them, and the backup must then restore.
+func TestPruneDuringBackupKeepsItsBlobs(t *testing.T) {
+	log.Initialize(log.Config{})
+	t.Parallel()
+
+	fs := afero.NewOsFs()
+	tempDir := t.TempDir()
+	dataDir := filepath.Join(tempDir, "src")
+	storeDir := t.TempDir()
+	restoreDir := filepath.Join(tempDir, "restored")
+	dbPath := filepath.Join(tempDir, "index.sqlite")
+
+	ctx := context.Background()
+	cfg := faultTestConfig()
+	testFiles := writeFaultSourceTree(t, fs, dataDir)
+
+	inner, err := storage.NewFileStorer(storeDir)
+	require.NoError(t, err)
+
+	otherHostCfg := faultTestConfig()
+	otherHostCfg.PruneGracePeriod = time.Hour
+	otherHost := newReaderVaultik(ctx, otherHostCfg, inner, nil, fs)
+
+	var (
+		pruned   bool
+		pruneErr error
+	)
+
+	store := faultstore.New(inner)
+	store.OnPut = func(key string) faultstore.PutAction {
+		if strings.HasSuffix(key, "/manifest.json.zst") {
+			pruned = true
+			pruneErr = otherHost.PruneBlobs(&vaultik.PruneOptions{Force: true})
+		}
+
+		return faultstore.PutNormal
+	}
+
+	db, err := database.New(ctx, dbPath)
+	require.NoError(t, err)
+
+	repos := database.NewRepositories(db)
+
+	id := fullFaultBackup(ctx, t, fs, store, cfg, repos, dataDir, dbPath, "shared")
+	require.NoError(t, db.Close())
+
+	require.True(t, pruned, "the prune must run before the manifest upload")
+	require.NoError(t, pruneErr)
+
+	v := newReaderVaultik(ctx, cfg, inner, nil, fs)
+	require.NoError(t, v.Restore(&vaultik.RestoreOptions{
+		SnapshotID: id,
+		TargetDir:  restoreDir,
+		Verify:     true,
+	}), "a prune during the backup must not delete the backup's blobs")
+
+	assertRestoredTree(t, fs, restoreDir, testFiles)
+}
+
+// TestPruneKeepsUnreferencedBlobWithNoTime checks that a blob no
+// manifest lists, whose listing gives no modification time (as an S3
+// listing entry without LastModified does), is kept and counted with
+// the blobs too new to delete: its age is unknown.
+func TestPruneKeepsUnreferencedBlobWithNoTime(t *testing.T) {
+	log.Initialize(log.Config{})
+	t.Parallel()
+
+	env := newListEnv(t)
+
+	// Written into the mock store's map directly rather than through
+	// Put, so its listing gives no time.
+	blobKey := "blobs/" + testBlobHashA[:2] + "/" + testBlobHashA[2:4] +
+		"/" + testBlobHashA
+	env.store.data[blobKey] = []byte("blob")
+
+	require.NoError(t, env.v.PruneBlobs(&vaultik.PruneOptions{JSON: true}))
+
+	assert.True(t, env.store.hasKey(blobKey),
+		"a blob of unknown age must be kept")
+
+	var result vaultik.PruneBlobsResult
+
+	require.NoError(t, json.Unmarshal(env.stdout.Bytes(), &result))
+	assert.Equal(t, 0, result.BlobsDeleted)
+	assert.Equal(t, 1, result.BlobsTooNew)
+}
+
+// TestPruneDeletesUnreferencedBlobOlderThanGracePeriod checks that the
+// grace period only delays deletion: a blob no manifest lists is deleted
+// once it is older than the grace period, and kept and counted until
+// then.
+func TestPruneDeletesUnreferencedBlobOlderThanGracePeriod(t *testing.T) {
+	log.Initialize(log.Config{})
+	t.Parallel()
+
+	ctx := context.Background()
+	storeDir := t.TempDir()
+
+	store, err := storage.NewFileStorer(storeDir)
+	require.NoError(t, err)
+
+	oldKey := "blobs/" + testBlobHashA[:2] + "/" + testBlobHashA[2:4] +
+		"/" + testBlobHashA
+	newKey := "blobs/" + testBlobHashB[:2] + "/" + testBlobHashB[2:4] +
+		"/" + testBlobHashB
+
+	for _, key := range []string{oldKey, newKey} {
+		require.NoError(t, store.Put(ctx, key, strings.NewReader("blob")))
+	}
+
+	gracePeriod := time.Hour
+	beforeGracePeriod := time.Now().Add(-gracePeriod - time.Minute)
+	require.NoError(t, os.Chtimes(filepath.Join(storeDir, oldKey),
+		beforeGracePeriod, beforeGracePeriod))
+
+	stdout := &bytes.Buffer{}
+	v := &vaultik.Vaultik{
+		Config:  &config.Config{PruneGracePeriod: gracePeriod},
+		Storage: store,
+		Stdout:  stdout,
+		UI:      ui.NewWithColor(stdout, false),
+	}
+	v.SetContext(ctx)
+
+	require.NoError(t, v.PruneBlobs(&vaultik.PruneOptions{JSON: true}))
+
+	_, err = store.Stat(ctx, oldKey)
+	require.ErrorIs(t, err, storage.ErrNotFound,
+		"a blob older than the grace period must be deleted")
+
+	_, err = store.Stat(ctx, newKey)
+	require.NoError(t, err, "a blob newer than the grace period must be kept")
+
+	var result vaultik.PruneBlobsResult
+
+	require.NoError(t, json.Unmarshal(stdout.Bytes(), &result))
+	assert.Equal(t, 1, result.BlobsDeleted)
+	assert.Equal(t, 1, result.BlobsTooNew)
+}

internal/vaultik/remove_snapshot_test.go

@@ -26,11 +26,14 @@ import (
 type testStorer struct {
 	mu   sync.Mutex
 	data map[string][]byte
+	// putTimes holds when Put stored each object; ListStream reports it.
+	putTimes map[string]time.Time
 }
 
 func newTestStorer() *testStorer {
 	return &testStorer{
-		data: make(map[string][]byte),
+		data:     make(map[string][]byte),
+		putTimes: make(map[string]time.Time),
 	}
 }
 
@@ -44,6 +47,7 @@ func (s *testStorer) Put(_ context.Context, key string, reader io.Reader) error
 	}
 
 	s.data[key] = data
+	s.putTimes[key] = time.Now().UTC()
 
 	return nil
 }
@@ -87,6 +91,7 @@ func (s *testStorer) Delete(_ context.Context, key string) error {
 	defer s.mu.Unlock()
 
 	delete(s.data, key)
+	delete(s.putTimes, key)
 
 	return nil
 }
@@ -120,8 +125,9 @@ func (s *testStorer) ListStream(
 		for key, data := range s.data {
 			if prefix == "" || strings.HasPrefix(key, prefix) {
 				ch <- storage.ObjectInfo{
-					Key:  key,
-					Size: int64(len(data)),
+					Key:     key,
+					Size:    int64(len(data)),
+					ModTime: s.putTimes[key],
 				}
 			}
 		}