fix(core): identify Kopia snapshots by a stable identity in verify and delete

AGENTS.md

@@ -204,8 +204,8 @@ confirm after that warning.
   canonical case: `kopia snapshot pin` rewrites the snapshot manifest to a *new*
   ID and deletes the old one, so without a refresh the server keeps serving the
   now-deleted ID for a backup that still exists, and a later client
-  `klio backup delete` deletes the wrong ID — leaving the real backup (and its
-  WALs) pinned forever.
+  `klio backup delete` asks Kopia to delete an ID that no longer matches
+  anything: the command fails and the real backup (and its WALs) stay pinned.
 - A direct write that only **deletes** snapshots (the tier1/tier2 retention
   apply) does **not** need a refresh: it removes IDs the server may still list,
   but it never rewrites a live backup's ID, and WAL retention is recomputed from
@@ -218,6 +218,34 @@ Do not introduce direct-write paths anywhere else. If, after warning the user, a
 new direct write is genuinely unavoidable, it must be paired with a server
 refresh of the affected tier.
 
+### Snapshot identity: manifest ID vs root object ID
+
+A snapshot's **manifest ID is not a stable identity**. `kopia snapshot pin`
+(the tier1 unpin above) rewrites a snapshot's manifest under a new ID and
+deletes the old one, so any code that lists snapshots and then acts on them a
+moment later can be holding an ID that no longer exists. Pick the identity by
+what the operation does:
+
+- **Reads that must survive a concurrent rewrite** use the root object ID
+  (`Manifest.RootEntry.ObjID`), which the rewrite leaves untouched. Backup
+  verification (`core/internal/client/klioclient/kopia/verify.go`) does this,
+  passing every root to Kopia as `--file-id` regardless of whether it is a
+  directory root (pgdata, metadata) or a file root (the control data file,
+  snapshotted on its own).
+- **Deletions must NOT use the root object ID.** Unchanged content dedupes to
+  the same root across backups (two backups of an idle tablespace share one), and
+  `kopia snapshot delete` removes *every* snapshot matching the ID it is given,
+  so deleting one backup by root ID can take another backup's snapshot with it.
+  Delete by manifest ID, and on failure re-list and retry so a concurrent
+  rewrite is picked up (`DeleteBackup` in the same package).
+- **The tier1 unpin is a write, not a read, and knowingly accepts the same
+  collision as delete.** The consumer's `getPinnedSnapshots`/`maintainTier2`
+  (`core/internal/consumer/backup.go`) also targets the root object ID, so a
+  root shared with another backup gets unpinned too. This is tolerated only
+  because the step is best-effort and the affected snapshot would be unpinned
+  anyway on the next tier2 migration — it is not a safe pattern to copy for
+  anything that isn't equally tolerant of that collision.
+
 ### Dagger caching issues
 
 When running e2e tests, Dagger may cache Helm repo indexes. If a new version of

core/internal/client/klioclient/kopia/delete.go

@@ -23,46 +23,96 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
+	"time"
 
 	"github.com/cloudnative-pg/machinery/pkg/log"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/cloudnative-pg/klio/core/internal/client/klioclient"
+	"github.com/cloudnative-pg/klio/core/internal/kopia"
 )
 
 // ErrBackupNotFound is returned when attempting to delete a backup that does not exist.
 var ErrBackupNotFound = errors.New("backup not found")
 
+// deleteBackupBackoff defines the backoff between failed attempts to delete snapshots of a backup.
+//
+//nolint:gochecknoglobals
+var deleteBackupBackoff = wait.Backoff{
+	Duration: time.Second,
+	Factor:   2,
+	Cap:      2 * time.Second,
+	Steps:    3,
+}
+
+// snapshotStore is the subset of the Kopia client that DeleteBackup needs.
+type snapshotStore interface {
+	ListSnapshots(ctx context.Context, tags map[string]string, logFn kopia.LogFunc) ([]kopia.Manifest, error)
+	DeleteSnapshot(ctx context.Context, id string) error
+}
+
 // DeleteBackup removes all snapshots associated with the backup with the provided name.
 func (s *Connection) DeleteBackup(ctx context.Context, hostname string, name string) error {
+	return deleteBackupSnapshots(ctx, s.kopia, hostname, name)
+}
+
+// deleteBackupSnapshots removes every snapshot of a backup on the given host.
+func deleteBackupSnapshots(ctx context.Context, store snapshotStore, hostname, name string) error {
 	contextLogger := log.FromContext(ctx)
 
-	// List all snapshots for this backup (all content types)
-	entries, err := s.kopia.ListSnapshots(ctx, map[string]string{
-		klioclient.BackupNameTagName: name,
-	}, contextLogger.Debug)
-	if err != nil {
-		return fmt.Errorf("while listing snapshots: %w", err)
-	}
+	// Concurrent operations on the same snapshots (e.g. post-backup
+	// maintenance) can invalidate the IDs resolved below before they are
+	// used, so this race condition is tolerated with retries.
+	retryErr := wait.ExponentialBackoffWithContext(ctx, deleteBackupBackoff,
+		func(ctx context.Context) (bool, error) {
+			entries, err := store.ListSnapshots(ctx, map[string]string{
+				klioclient.BackupNameTagName: name,
+			}, contextLogger.Debug)
+			if err != nil {
+				return false, fmt.Errorf("while listing snapshots: %w", err)
+			}
 
-	var deleted int
-	for _, entry := range entries {
-		if entry.Source.Host == hostname {
-			contextLogger.Info("DeleteBackup: deleting snapshot", "snapshotID", entry.ID)
-			if deleteErr := s.kopia.DeleteSnapshot(ctx, entry.ID); deleteErr != nil {
-				err = errors.Join(err, deleteErr)
-			} else {
-				deleted++
+			entries = slices.DeleteFunc(entries, func(e kopia.Manifest) bool {
+				return e.Source.Host != hostname
+			})
+
+			if len(entries) == 0 {
+				return false, fmt.Errorf("%w: %s", ErrBackupNotFound, name)
 			}
-		}
-	}
 
-	if err != nil {
-		return err
-	}
+			err = deleteSnapshots(ctx, store, entries)
+			if err == nil {
+				return true, nil
+			}
+
+			contextLogger.Info("DeleteBackup: an error occurred while trying to delete backup's snapshots, retrying...",
+				"backupName", name, "error", err)
+
+			return false, nil
+		},
+	)
+
+	return retryErr
+}
 
-	if deleted == 0 {
-		return fmt.Errorf("%w: %s", ErrBackupNotFound, name)
+// deleteSnapshots deletes every given entry, joining and returning any
+// deletion errors.
+func deleteSnapshots(
+	ctx context.Context,
+	store snapshotStore,
+	entries []kopia.Manifest,
+) error {
+	contextLogger := log.FromContext(ctx)
+
+	var err error
+
+	for _, entry := range entries {
+		contextLogger.Info("DeleteBackup: deleting snapshot", "snapshotID", entry.ID)
+		if deleteErr := store.DeleteSnapshot(ctx, entry.ID); deleteErr != nil {
+			err = errors.Join(err, deleteErr)
+		}
 	}
 
-	return nil
+	return err
 }

core/internal/client/klioclient/kopia/delete_test.go

@@ -0,0 +1,158 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package kopia
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/cloudnative-pg/klio/core/internal/kopia"
+)
+
+var errDeleteFailed = errors.New("no snapshots matched")
+
+// fakeSnapshotStore serves a scripted sequence of snapshot listings and records
+// the IDs it was asked to delete. Listings are consumed one per call so a test
+// can model the manifest IDs changing between attempts.
+type fakeSnapshotStore struct {
+	listings  [][]kopia.Manifest
+	listCalls int
+
+	// failDeleteOf holds the IDs whose deletion fails.
+	failDeleteOf map[string]bool
+
+	deleted []string
+	listErr error
+}
+
+func (f *fakeSnapshotStore) ListSnapshots(
+	_ context.Context,
+	_ map[string]string,
+	_ kopia.LogFunc,
+) ([]kopia.Manifest, error) {
+	if f.listErr != nil {
+		return nil, f.listErr
+	}
+
+	f.listCalls++
+
+	// The last scripted listing is reused for any further attempt.
+	idx := min(f.listCalls-1, len(f.listings)-1)
+
+	return f.listings[idx], nil
+}
+
+func (f *fakeSnapshotStore) DeleteSnapshot(_ context.Context, id string) error {
+	if f.failDeleteOf[id] {
+		return errDeleteFailed
+	}
+
+	f.deleted = append(f.deleted, id)
+
+	return nil
+}
+
+func manifest(id, host string) kopia.Manifest {
+	return kopia.Manifest{ID: id, Source: kopia.SourceInfo{Host: host}}
+}
+
+func TestDeleteBackupSnapshots(t *testing.T) {
+	ctx := context.Background()
+
+	oldBackoff := deleteBackupBackoff
+	deleteBackupBackoff.Duration = 0
+	deleteBackupBackoff.Cap = 0
+	t.Cleanup(func() { deleteBackupBackoff = oldBackoff })
+
+	t.Run("deletes every snapshot of the backup on the host", func(t *testing.T) {
+		store := &fakeSnapshotStore{
+			listings: [][]kopia.Manifest{
+				{manifest("a", "cluster"), manifest("b", "cluster")},
+				{},
+			},
+		}
+
+		require.NoError(t, deleteBackupSnapshots(ctx, store, "cluster", "backup-1"))
+		assert.Equal(t, []string{"a", "b"}, store.deleted)
+		assert.Equal(t, 1, store.listCalls)
+	})
+
+	t.Run("ignores snapshots belonging to another host", func(t *testing.T) {
+		store := &fakeSnapshotStore{
+			listings: [][]kopia.Manifest{{manifest("a", "other-cluster")}},
+		}
+
+		err := deleteBackupSnapshots(ctx, store, "cluster", "backup-1")
+
+		require.ErrorIs(t, err, ErrBackupNotFound)
+		assert.Empty(t, store.deleted)
+	})
+
+	t.Run("returns not found when the backup has no snapshots", func(t *testing.T) {
+		store := &fakeSnapshotStore{listings: [][]kopia.Manifest{{}}}
+
+		require.ErrorIs(t, deleteBackupSnapshots(ctx, store, "cluster", "backup-1"), ErrBackupNotFound)
+	})
+
+	// A concurrent "kopia snapshot pin" rewrites a snapshot's manifest under a
+	// new ID, so deleting the ID we listed matches nothing. Re-resolving must
+	// pick up the new ID and finish the deletion.
+	t.Run("retries with the rewritten manifest ID", func(t *testing.T) {
+		store := &fakeSnapshotStore{
+			listings: [][]kopia.Manifest{
+				{manifest("stale", "cluster")},
+				{manifest("stale", "cluster")},
+				{manifest("rewritten", "cluster")},
+				{},
+			},
+			failDeleteOf: map[string]bool{"stale": true},
+		}
+
+		require.NoError(t, deleteBackupSnapshots(ctx, store, "cluster", "backup-1"))
+		assert.Equal(t, []string{"rewritten"}, store.deleted)
+		assert.Equal(t, 3, store.listCalls)
+	})
+
+	t.Run("gives up after the attempt budget and reports the failure", func(t *testing.T) {
+		store := &fakeSnapshotStore{
+			listings:     [][]kopia.Manifest{{manifest("stuck", "cluster")}},
+			failDeleteOf: map[string]bool{"stuck": true},
+		}
+
+		err := deleteBackupSnapshots(ctx, store, "cluster", "backup-1")
+
+		require.Error(t, err)
+		assert.True(t, wait.Interrupted(err), "expected the attempt budget to be exhausted, got: %v", err)
+		assert.Equal(t, deleteBackupBackoff.Steps, store.listCalls)
+		assert.Empty(t, store.deleted)
+	})
+
+	t.Run("a listing failure is returned as is", func(t *testing.T) {
+		listErr := errors.New("connection refused")
+		store := &fakeSnapshotStore{listErr: listErr}
+
+		require.ErrorIs(t, deleteBackupSnapshots(ctx, store, "cluster", "backup-1"), listErr)
+	})
+}