@@ -38,6 +38,13 @@ type s3ObjectLister interface {
 	ListObjects(ctx context.Context, bucket, prefix string, recursive bool) ([]s3ObjectInfo, error)
 }
 
+type s3ObjectRemover interface {
+	// RemoveObject deletes key. S3 answers 204 for a key that does not
+	// exist and minio-go takes 204 as success; a backend answering a
+	// missing key with any other status makes it an error.
+	RemoveObject(ctx context.Context, bucket, key string) error
+}
+
 // s3Connection is what a driver resolves from its rendered strategy to
 // read the bucket with the same identity the backup was written with.
 type s3Connection struct {
@@ -63,8 +70,10 @@ type objectStore struct {
 	// reader reads Secrets, which the manager's client fetches uncached,
 	// and lists operator Backups, which it serves from the cache.
 	reader client.Reader
-	// newLister overrides the S3 client; nil means a real minio client.
-	newLister func(conn s3Connection) (s3ObjectLister, error)
+	// newLister and newRemover override the S3 client; nil means a real
+	// minio client.
+	newLister  func(conn s3Connection) (s3ObjectLister, error)
+	newRemover func(conn s3Connection) (s3ObjectRemover, error)
 }
 
 func (s objectStore) lister(conn s3Connection) (s3ObjectLister, error) {
@@ -74,6 +83,17 @@ func (s objectStore) lister(conn s3Connection) (s3ObjectLister, error) {
 	return newMinioObjectLister(conn)
 }
 
+func (s objectStore) remover(conn s3Connection) (s3ObjectRemover, error) {
+	if s.newRemover != nil {
+		return s.newRemover(conn)
+	}
+	c, err := newMinioObjectClient(conn)
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
 func (s objectStore) secretValue(ctx context.Context, namespace, name, key string) (string, error) {
 	secret := &corev1.Secret{}
 	if err := s.reader.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, secret); err != nil {
@@ -94,6 +114,10 @@ func (r *RestoreJobReconciler) objectStore() objectStore {
 	return objectStore{reader: r.Client, newLister: r.newS3ObjectLister}
 }
 
+func (r *BackupReconciler) objectStore() objectStore {
+	return objectStore{reader: r.Client, newLister: r.newS3ObjectLister, newRemover: r.newS3ObjectRemover}
+}
+
 func (r *BackupJobReconciler) s3Lister(conn s3Connection) (s3ObjectLister, error) {
 	return r.objectStore().lister(conn)
 }
@@ -103,6 +127,14 @@ type minioObjectLister struct {
 }
 
 func newMinioObjectLister(conn s3Connection) (s3ObjectLister, error) {
+	c, err := newMinioObjectClient(conn)
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
+func newMinioObjectClient(conn s3Connection) (*minioObjectLister, error) {
 	opts := &minio.Options{
 		Creds:  credentials.NewStaticV4(conn.AccessKey, conn.SecretKey, conn.SessionToken),
 		Region: conn.Region,
@@ -152,6 +184,10 @@ func (l *minioObjectLister) ListObjects(ctx context.Context, bucket, prefix stri
 	return collectObjects(ctx, l.client.ListObjects(ctx, bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: recursive}))
 }
 
+func (l *minioObjectLister) RemoveObject(ctx context.Context, bucket, key string) error {
+	return l.client.RemoveObject(ctx, bucket, key, minio.RemoveObjectOptions{})
+}
+
 // collectObjects drains a minio listing. minio-go ends a listing whose
 // context is cancelled between pages by closing the channel without an
 // error item, so a cancelled context is checked once the channel closes:
 
@@ -39,9 +39,22 @@ type BackupReconciler struct {
 	client.Client
 	Scheme   *runtime.Scheme
 	Recorder record.EventRecorder
-	// CredentialsConfig lets the Rabbitmq cleanup path project cozy-backups-creds
-	// into the Backup's namespace before spawning its best-effort delete Job.
+	// CredentialsConfig locates the platform backup credentials: the Job-based
+	// cleanups project them into the Backup's namespace, and the MariaDB
+	// cleanup reads them to reach the platform bucket.
 	CredentialsConfig BackupCredentialsConfig
+	// APIReader is the manager's uncached reader. The MariaDB cleanup lists
+	// the RestoreJobs and operator Restores of a Backup, and the operator
+	// Backups that may own its dump, through it: a restore that started
+	// before the delete, or a run that has already uploaded its dump, may
+	// not be in the cache yet. Wired in SetupWithManager.
+	APIReader client.Reader
+
+	// newS3ObjectLister and newS3ObjectRemover override the S3 client the
+	// MariaDB cleanup deletes a dump with; nil means a real minio client.
+	// Set by unit tests.
+	newS3ObjectLister  func(conn s3Connection) (s3ObjectLister, error)
+	newS3ObjectRemover func(conn s3Connection) (s3ObjectRemover, error)
 }
 
 func (r *BackupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -126,22 +139,20 @@ func (r *BackupReconciler) cleanupOnDelete(ctx context.Context, backup *backupsv
 		// orchestration" for tenant-facing guidance.
 		return ctrl.Result{}, nil
 	case strategyv1alpha1.MariaDBStrategyKind:
-		// Cozystack Backup deletion does NOT delete the operator-side
-		// k8s.mariadb.com/Backup CR or the backing S3/PVC archive.
-		// Lifecycle of that CR is owned by mariadb-operator's
-		// MaxRetention setting and by tenants / explicit teardown flows;
-		// the controller's RBAC has no delete verb on
-		// k8s.mariadb.com/backups for that reason. Surfacing this as an
-		// explicit branch (rather than falling through to the Velero
-		// default) keeps the contract honest: the MariaDB driver does
-		// not own the archive, so it does not delete it on CR removal.
-		// Mirrors the Altinity branch's "we do not own S3" stance.
-		return ctrl.Result{}, nil
+		// mariadb-operator's maxRetention prunes only from inside the next
+		// backup run of the same prefix and a Plan never deletes a Backup, so
+		// a dump on the platform bucket is deleted here or never. A dump in
+		// a tenant's own bucket is deleted too: the Backup proves it owns
+		// it, and tenants cannot delete Backups, so the delete comes from an
+		// administrator or from namespace teardown. Teardown can remove the
+		// operator Backup first, and the dump is then left behind with a
+		// Warning, since its storage is read from it (see cleanupMariaDBBackup).
+		return r.cleanupMariaDBBackup(ctx, backup)
 	case strategyv1alpha1.MongoDBStrategyKind:
 		// Split by flow (see cleanupMongoDBBackup). A legacy backup writes to the
 		// tenant's own bucket and its lifecycle stays the tenant's / psmdb
 		// operator's — that branch is a no-op, the same "we do not own the
-		// archive" contract as Altinity / MariaDB / FDB. A useSystemBucket backup
+		// archive" contract as Altinity / FDB. A useSystemBucket backup
 		// lands in the shared cozy-backups bucket with no psmdb task retention to
 		// prune it, so the driver OWNS it: this deletes the operator
 		// PerconaServerMongoDBBackup CR and waits for its delete-backup finalizer
@@ -154,7 +165,7 @@ func (r *BackupReconciler) cleanupOnDelete(ctx context.Context, backup *backupsv
 		// is governed by the next BackupJob's stopOtherFoundationDBBackups
 		// call, by manual operator action, or by
 		// examples/backups/foundationdb/cleanup.sh. Same "we do not own
-		// the archive" contract as Altinity / MariaDB — the explicit
+		// the archive" contract as Altinity — the explicit
 		// branch guards the seam against a future driver refactor that
 		// might incidentally stamp velero.io/backup-name onto FDB
 		// driverMetadata via a shared helper.
@@ -291,7 +302,18 @@ func (r *BackupReconciler) cleanupCNPGBackup(ctx context.Context, backup *backup
 
 // SetupWithManager registers the BackupReconciler with the Manager.
 func (r *BackupReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.APIReader = mgr.GetAPIReader()
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&backupsv1alpha1.Backup{}).
 		Complete(r)
 }
+
+// apiReader returns the uncached reader. Reconcilers built by hand in unit
+// tests have none and fall back to their fake Client, which has no cache to
+// bypass.
+func (r *BackupReconciler) apiReader() client.Reader {
+	if r.APIReader != nil {
+		return r.APIReader
+	}
+	return r.Client
+}
@@ -176,11 +176,8 @@ func TestBackupCleanup_Altinity_DoesNotFallThroughToVelero(t *testing.T) {
 
 // TestBackupCleanup_MariaDB_DoesNotFallThroughToVelero locks in the same
 // invariant as the Altinity test above for the MariaDB strategy. The
-// dispatcher must explicitly route MariaDB-strategy Backups to a no-op
-// cleanup rather than fall through to the Velero default - the MariaDB
-// driver does not own the operator-side k8s.mariadb.com/Backup CR or the
-// S3/PVC archive behind it (rbac.yaml grants no delete verb on
-// k8s.mariadb.com/backups for that reason). The test seeds the Velero
+// dispatcher must route MariaDB-strategy Backups to their own cleanup
+// rather than fall through to the Velero default. The test seeds the Velero
 // metadata keys plus a matching velero.io/Backup; with the explicit branch
 // no DeleteBackupRequest is synthesised, and the Velero Backup survives.
 // Without the branch the default falls through to cleanupVeleroBackup,
 
@@ -60,11 +60,23 @@ func (r *BackupJobReconciler) attributedMariaDBDump(ctx context.Context, mdbBack
 // from can have been edited since, and pointing at another bucket now says
 // nothing about where this dump went.
 func findMariaDBDump(ctx context.Context, store objectStore, mdbBackup *mariadbtypes.Backup, completedAt time.Time) (mariadbDump, error) {
-	lister, s3, err := mariadbStorage(ctx, store, mdbBackup)
+	lister, _, err := mariadbStorage(ctx, store, mdbBackup)
 	if err != nil {
 		return mariadbDump{}, err
 	}
-	others, err := finishedMariaDBRuns(ctx, store.reader, mdbBackup)
+	return findMariaDBDumpWith(ctx, lister, store.reader, mdbBackup, completedAt, false)
+}
+
+// findMariaDBDumpWith is findMariaDBDump reading the bucket through lister,
+// for a caller that reaches the bucket by another endpoint than the one
+// mdbBackup names. withUnfinished lets runs that have not completed contest
+// the dump too, as neighbourMariaDBRuns describes.
+func findMariaDBDumpWith(ctx context.Context, lister s3ObjectLister, reader client.Reader, mdbBackup *mariadbtypes.Backup, completedAt time.Time, withUnfinished bool) (mariadbDump, error) {
+	s3 := mdbBackup.Spec.Storage.S3
+	if s3 == nil {
+		return mariadbDump{}, errors.New("the operator Backup does not store to S3")
+	}
+	others, err := neighbourMariaDBRuns(ctx, reader, mdbBackup, withUnfinished)
 	if err != nil {
 		return mariadbDump{}, fmt.Errorf("list operator Backups: %w", err)
 	}
@@ -125,6 +137,10 @@ type mariadbRun struct {
 	lower, upper time.Time
 }
 
+// mariadbUnfinishedRunEnd closes the window of a run that has not completed,
+// which can still write its dump at any time.
+var mariadbUnfinishedRunEnd = time.Date(9999, time.December, 31, 0, 0, 0, 0, time.UTC)
+
 func newMariaDBRun(b *mariadbtypes.Backup, created, completed time.Time) mariadbRun {
 	return mariadbRun{
 		name:  b.Namespace + "/" + b.Name,
@@ -150,8 +166,8 @@ func (r mariadbRun) placesOnEitherClock(d mariadbDump) bool {
 // dump in the window on one clock only may be this run's own, misplaced by
 // a skewed clock, which leaves any other candidate unproven. A neighbour's
 // dump is still taken if this run's own is missing from the listing, or
-// misplaced on both clocks, and the neighbour's operator Backup is gone or
-// has not completed.
+// misplaced on both clocks, and the neighbour's operator Backup is gone or,
+// unless others includes unfinished runs, has not completed.
 func matchMariaDBDump(objects []s3ObjectInfo, prefix string, own mariadbRun, others []mariadbRun) (mariadbDump, error) {
 	var near []mariadbDump
 	for _, obj := range objects {
@@ -183,14 +199,17 @@ func matchMariaDBDump(objects []s3ObjectInfo, prefix string, own mariadbRun, oth
 	return d, nil
 }
 
-// finishedMariaDBRuns returns the runs of the other operator Backups that
+// neighbourMariaDBRuns returns the runs of the other operator Backups that
 // write under the same bucket and prefix as b and have completed, whether
-// their Job succeeded or failed. One still running is left out: its window
-// has no end yet, and an operator Backup stuck short of Complete, as one
-// whose Job pod never starts is, would otherwise contest every later run of
-// the application for good. The endpoint is not compared, since one has
-// several spellings, and a false neighbour only leaves a size unset.
-func finishedMariaDBRuns(ctx context.Context, reader client.Reader, b *mariadbtypes.Backup) ([]mariadbRun, error) {
+// their Job succeeded or failed. One still running is left out unless
+// withUnfinished is set: its window has no end yet, and an operator Backup
+// stuck short of Complete, as one whose Job pod never starts is, would
+// otherwise contest every later run of the application for good. That is
+// the price of deleting a dump, since a run can upload its dump and prune
+// older ones through maxRetention before it completes. The endpoint is not
+// compared, since one has several spellings, and a false neighbour only
+// keeps a dump unattributed.
+func neighbourMariaDBRuns(ctx context.Context, reader client.Reader, b *mariadbtypes.Backup, withUnfinished bool) ([]mariadbRun, error) {
 	list := &mariadbtypes.BackupList{}
 	if err := reader.List(ctx, list); err != nil {
 		return nil, err
@@ -207,6 +226,9 @@ func finishedMariaDBRuns(ctx context.Context, reader client.Reader, b *mariadbty
 		}
 		cond := apimeta.FindStatusCondition(other.Status.Conditions, mariadbtypes.ConditionTypeComplete)
 		if cond == nil || cond.Status != metav1.ConditionTrue {
+			if withUnfinished {
+				runs = append(runs, newMariaDBRun(other, other.CreationTimestamp.Time, mariadbUnfinishedRunEnd))
+			}
 			continue
 		}
 		runs = append(runs, newMariaDBRun(other, other.CreationTimestamp.Time, cond.LastTransitionTime.Time))
 
@@ -14,6 +14,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
 
 	backupsv1alpha1 "github.com/cozystack/cozystack/api/backups/v1alpha1"
 	"github.com/cozystack/cozystack/internal/backupcontroller/mariadbapp"
@@ -184,6 +185,117 @@ func recordedDump(name string) map[string]string {
 	return map[string]string{mariadbDumpFileKey: name, mariadbDumpTimeKey: mdbRestoreTime}
 }
 
+// A Backup being deleted is deleting its dump, so a restore that starts
+// now can find it gone once the operator reads it, and the operator then
+// restores the nearest other dump. The Backup is read from the API server,
+// since the copy the reconcile was handed can predate the delete.
+func TestReconcileMariaDBRestore_RefusesABackupBeingDeleted(t *testing.T) {
+	creds := s3CredsSecret("tenant", "creds")
+	present := func() *fakeS3Lister {
+		return &fakeS3Lister{objects: []s3ObjectInfo{{Key: mdbRestorePrefix + mdbRestoreDump, LastModified: mdbRestoreT0.Add(20 * time.Second)}}}
+	}
+	cases := map[string]map[string]string{"recorded dump": recordedDump(mdbRestoreDump), "legacy Backup": nil}
+	for name, md := range cases {
+		t.Run(name+", being deleted", func(t *testing.T) {
+			f := newMariaDBRestoreFixture(t, md, present(), creds)
+			ctx := context.Background()
+			live := &backupsv1alpha1.Backup{}
+			if err := f.c.Get(ctx, client.ObjectKeyFromObject(f.backup), live); err != nil {
+				t.Fatal(err)
+			}
+			live.Finalizers = []string{backupFinalizer}
+			if err := f.c.Update(ctx, live); err != nil {
+				t.Fatal(err)
+			}
+			if err := f.c.Delete(ctx, live); err != nil {
+				t.Fatal(err)
+			}
+			f.reconcile(t)
+			f.wantFailed(t, "src-bk", "being deleted")
+		})
+		t.Run(name+", already gone", func(t *testing.T) {
+			f := newMariaDBRestoreFixture(t, md, present(), creds)
+			if err := f.c.Delete(context.Background(), f.backup.DeepCopy()); err != nil {
+				t.Fatal(err)
+			}
+			f.reconcile(t)
+			f.wantFailed(t, "src-bk", "is gone")
+		})
+	}
+	// The refusal applies only before the operator Restore exists. One
+	// created by an earlier reconcile but not yet in the cache must be
+	// found and followed, or the RestoreJob fails under a running restore
+	// and the cleanup deletes the dump that restore reads.
+	t.Run("an operator Restore the cache has not seen yet is followed", func(t *testing.T) {
+		f := newMariaDBRestoreFixture(t, recordedDump(mdbRestoreDump), present(), creds)
+		ctx := context.Background()
+		live := &backupsv1alpha1.Backup{}
+		if err := f.c.Get(ctx, client.ObjectKeyFromObject(f.backup), live); err != nil {
+			t.Fatal(err)
+		}
+		live.Finalizers = []string{backupFinalizer}
+		if err := f.c.Update(ctx, live); err != nil {
+			t.Fatal(err)
+		}
+		if err := f.c.Delete(ctx, live); err != nil {
+			t.Fatal(err)
+		}
+		if err := f.c.Get(ctx, client.ObjectKeyFromObject(live), live); err != nil || live.DeletionTimestamp.IsZero() {
+			t.Fatalf("Backup not left being deleted: %v", err)
+		}
+		live.ResourceVersion = ""
+		apiServer := newMariaDBStrategyTestClient(t, live, &mariadbtypes.Restore{ObjectMeta: metav1.ObjectMeta{
+			Namespace: "tenant", Name: "rj-abcde",
+			Labels: map[string]string{
+				backupsv1alpha1.OwningJobNameLabel:      "rj",
+				backupsv1alpha1.OwningJobNamespaceLabel: "tenant",
+			},
+		}})
+		f.r.APIReader = apiServer
+		f.reconcile(t)
+		if f.rj.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
+			t.Fatalf("RestoreJob failed under a running restore: %s", f.rj.Status.Message)
+		}
+		if got := f.restores(t); len(got) != 0 {
+			t.Errorf("created %d operator Restores; want the existing one followed", len(got))
+		}
+	})
+	// A failed lookup says nothing about whether the operator Restore
+	// exists, so the RestoreJob must not be failed under one that runs.
+	t.Run("an operator Restore lookup that fails is retried", func(t *testing.T) {
+		f := newMariaDBRestoreFixture(t, recordedDump(mdbRestoreDump), present(), creds)
+		f.r.APIReader = interceptor.NewClient(f.c.(client.WithWatch), interceptor.Funcs{
+			List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
+				if _, ok := list.(*mariadbtypes.RestoreList); ok {
+					return errors.New("etcdserver: request timed out")
+				}
+				return c.List(ctx, list, opts...)
+			},
+		})
+		if _, err := f.r.reconcileMariaDBRestore(context.Background(), f.rj, f.backup); err == nil {
+			t.Error("err = nil, want the failed lookup returned for a retry")
+		}
+		if f.rj.Status.Phase == backupsv1alpha1.RestoreJobPhaseFailed {
+			t.Fatalf("RestoreJob failed on a lookup error: %s", f.rj.Status.Message)
+		}
+		if got := f.restores(t); len(got) != 0 {
+			t.Errorf("created %d operator Restores without knowing whether one exists", len(got))
+		}
+	})
+	t.Run("a Backup that cannot be read keeps the RestoreJob waiting", func(t *testing.T) {
+		f := newMariaDBRestoreFixture(t, recordedDump(mdbRestoreDump), present(), creds)
+		f.r.APIReader = interceptor.NewClient(f.c.(client.WithWatch), interceptor.Funcs{
+			Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+				if _, ok := obj.(*backupsv1alpha1.Backup); ok {
+					return errors.New("apiserver unavailable")
+				}
+				return c.Get(ctx, key, obj, opts...)
+			},
+		})
+		f.wantWaiting(t, f.reconcile(t), "src-bk", "apiserver unavailable")
+	})
+}
+
 func TestReconcileMariaDBRestore_RecordedDump(t *testing.T) {
 	creds := s3CredsSecret("tenant", "creds")