fix(backups): fail a ClickHouse BackupJob whose Backup name is taken

A Backup outlives the BackupJob that made it, so a BackupJob created
again under a name already used meets the earlier run's Backup when its
own Job completes. That happens by hand, and from a Plan reconciled
again inside its starting deadline, since a Plan names each BackupJob
after its schedule slot. The Altinity driver took the earlier Backup as
its result: the BackupJob turned Succeeded with a BackupRef recording
the earlier run's archive, the archive this run wrote was recorded on
nothing, and restoring what the BackupJob reported brought back the
older data.

An existing Backup is now taken as this run's only when it records this
run's archive, which is what an earlier reconcile of the same run
leaves when its status write is lost. Any other fails the BackupJob,
naming both archives, and leaves the earlier Backup untouched.

Assisted-by: LLM
Signed-off-by: Myasnikov Daniil <myasnikovdaniil2001@gmail.com>

internal/backupcontroller/altinitystrategy_controller.go

@@ -509,6 +509,13 @@ func (r *BackupJobReconciler) createAltinityBackupArtifact(
 		if getErr := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
 			return nil, getErr
 		}
+		// A Backup outlives its BackupJob, so a BackupJob re-created under a
+		// used name meets the earlier run's Backup here. Only one recording
+		// this run's archive was created by an earlier reconcile of this run.
+		if got := existing.Spec.DriverMetadata[altinityArchiveMetaKey]; got != archive {
+			return nil, fmt.Errorf("Backup %s/%s already exists and records archive %q, not %q that this run wrote",
+				backup.Namespace, backup.Name, got, archive)
+		}
 		return existing, nil
 	}
 	return backup, nil

internal/backupcontroller/altinitystrategy_controller_test.go

@@ -3,6 +3,7 @@ package backupcontroller
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -861,6 +862,119 @@ func TestReconcileAltinity_RecordsNoArchiveTheJobWasNotGiven(t *testing.T) {
 	}
 }
 
+// TestReconcileAltinity_BackupNameAlreadyTaken covers a completed Job whose
+// Backup name is already in use. A Backup outlives its BackupJob, so a
+// BackupJob re-created under an earlier run's name finds that run's Backup,
+// and claiming it would point the new run at an archive it did not write.
+// A Backup recording this run's own archive is the one an earlier reconcile
+// created before its status write was lost, and is adopted.
+func TestReconcileAltinity_BackupNameAlreadyTaken(t *testing.T) {
+	const thisRun = "clickhouse-ch-test-20260305T050607Z"
+	cases := []struct {
+		name          string
+		existing      string // archive the existing Backup records; empty for none
+		wantSucceeded bool
+	}{
+		{name: "earlier run's Backup records another archive", existing: "clickhouse-ch-test-20260304T050607Z"},
+		{name: "earlier run's Backup records no archive", existing: ""},
+		{name: "this run's Backup from a reconcile whose status write was lost", existing: thisRun, wantSucceeded: true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			app := newAltinityClickHouseApp("ch-test", "tenant-test")
+			strategy := newAltinityArchiveStrategy("clickhouse-strategy")
+			startedAt := metav1.NewTime(time.Date(2026, 3, 5, 5, 6, 7, 0, time.UTC))
+			backupJob := &backupsv1alpha1.BackupJob{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-bj", Namespace: "tenant-test"},
+				Spec: backupsv1alpha1.BackupJobSpec{
+					ApplicationRef:  newAltinityClickHouseRef("ch-test"),
+					BackupClassName: "clickhouse-backup",
+				},
+				Status: backupsv1alpha1.BackupJobStatus{StartedAt: &startedAt, Phase: backupsv1alpha1.BackupJobPhaseRunning},
+			}
+			driverMD := map[string]string{altinityParamPrefix + "bucketName": "ch-bucket"}
+			if tc.existing != "" {
+				driverMD[altinityArchiveMetaKey] = tc.existing
+			}
+			existing := &backupsv1alpha1.Backup{
+				ObjectMeta: metav1.ObjectMeta{Name: backupJob.Name, Namespace: backupJob.Namespace},
+				Spec: backupsv1alpha1.BackupSpec{
+					ApplicationRef: newAltinityClickHouseRef("ch-test"),
+					StrategyRef: corev1.TypedLocalObjectReference{
+						APIGroup: new(strategyv1alpha1.GroupVersion.Group),
+						Kind:     strategyv1alpha1.AltinityStrategyKind,
+						Name:     "clickhouse-strategy",
+					},
+					TakenAt:        metav1.NewTime(time.Date(2026, 3, 4, 5, 6, 22, 0, time.UTC)),
+					DriverMetadata: driverMD,
+				},
+			}
+			resolved := &ResolvedBackupConfig{
+				StrategyRef: corev1.TypedLocalObjectReference{
+					APIGroup: new(strategyv1alpha1.GroupVersion.Group),
+					Kind:     strategyv1alpha1.AltinityStrategyKind,
+					Name:     "clickhouse-strategy",
+				},
+				Parameters: map[string]string{"bucketName": "ch-bucket"},
+			}
+			r, _, _ := newAltinityTestEnv(t, app,
+				clientfake.NewClientBuilder().WithObjects(backupJob, strategy, existing))
+			ctx := context.Background()
+
+			if _, err := r.reconcileAltinity(ctx, backupJob, resolved); err != nil {
+				t.Fatalf("reconcileAltinity() creating the Job: %v", err)
+			}
+			job := &batchv1.Job{}
+			if err := r.Get(ctx, client.ObjectKey{Namespace: "tenant-test", Name: jobNameForBackupJob(backupJob)}, job); err != nil {
+				t.Fatalf("get backup Job: %v", err)
+			}
+			if got := job.Spec.Template.Spec.Containers[0].Args[1]; got != "--archive="+thisRun {
+				t.Fatalf("backup Pod was rendered with %q, want --archive=%s", got, thisRun)
+			}
+			job.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}}
+			if err := r.Status().Update(ctx, job); err != nil {
+				t.Fatalf("mark backup Job complete: %v", err)
+			}
+			if err := r.Get(ctx, client.ObjectKeyFromObject(backupJob), backupJob); err != nil {
+				t.Fatalf("refresh BackupJob: %v", err)
+			}
+			if _, err := r.reconcileAltinity(ctx, backupJob, resolved); err != nil {
+				t.Fatalf("reconcileAltinity() on the completed Job: %v", err)
+			}
+
+			updated := &backupsv1alpha1.BackupJob{}
+			if err := r.Get(ctx, client.ObjectKeyFromObject(backupJob), updated); err != nil {
+				t.Fatalf("get BackupJob: %v", err)
+			}
+			if tc.wantSucceeded {
+				if updated.Status.Phase != backupsv1alpha1.BackupJobPhaseSucceeded {
+					t.Errorf("phase = %q (%s), want Succeeded", updated.Status.Phase, updated.Status.Message)
+				}
+				if updated.Status.BackupRef == nil || updated.Status.BackupRef.Name != existing.Name {
+					t.Errorf("BackupRef = %v, want %s", updated.Status.BackupRef, existing.Name)
+				}
+			} else {
+				if updated.Status.Phase != backupsv1alpha1.BackupJobPhaseFailed {
+					t.Errorf("phase = %q, want Failed: the run would claim a Backup recording %q instead of %q", updated.Status.Phase, tc.existing, thisRun)
+				}
+				if updated.Status.BackupRef != nil {
+					t.Errorf("BackupRef = %v, want none", updated.Status.BackupRef)
+				}
+				if !strings.Contains(updated.Status.Message, thisRun) {
+					t.Errorf("failure message %q does not name the archive this run wrote", updated.Status.Message)
+				}
+			}
+			kept := &backupsv1alpha1.Backup{}
+			if err := r.Get(ctx, client.ObjectKeyFromObject(existing), kept); err != nil {
+				t.Fatalf("get existing Backup: %v", err)
+			}
+			if got := kept.Spec.DriverMetadata[altinityArchiveMetaKey]; got != tc.existing {
+				t.Errorf("existing Backup now records archive %q, want %q left in place", got, tc.existing)
+			}
+		})
+	}
+}
+
 func TestReconcileAltinityRestore_RendersTheBackupsArchive(t *testing.T) {
 	cases := []struct {
 		name        string