fix: discover affected Roles by label instead of listing Clusters

The ObjectStore controller now lists Roles by a label
(barmancloud.cnpg.io/cluster) set by the Pre hook, inspects their
rules to find which ObjectStores they reference, then fetches those
ObjectStores and rebuilds the rules. This removes the clusters
get/list/watch permission. Conflict handling uses RetryOnConflict to
match the existing project pattern, and partial failures across Roles
are aggregated with errors.Join instead of failing on the first one.

Pre-existing Roles without the label won't be found by the ObjectStore
controller until the Pre hook adds it on the next Cluster
reconciliation. Same staleness window as the current main branch.

Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

config/rbac/role.yaml

@@ -44,7 +44,6 @@ rules:
   - postgresql.cnpg.io
   resources:
   - backups
-  - clusters
   verbs:
   - get
   - list

internal/cnpgi/metadata/constants.go

@@ -26,6 +26,10 @@ import "github.com/cloudnative-pg/cnpg-i/pkg/identity"
 const PluginName = "barman-cloud.cloudnative-pg.io"
 
 const (
+	// ClusterLabelName is the label applied to RBAC resources created
+	// by this plugin. Its value is the name of the owning Cluster.
+	ClusterLabelName = "barmancloud.cnpg.io/cluster"
+
 	// CheckEmptyWalArchiveFile is the name of the file in the PGDATA that,
 	// if present, requires the WAL archiver to check that the backup object
 	// store is empty.

internal/cnpgi/operator/rbac/ensure.go

@@ -29,10 +29,12 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/specs"
 )
 
@@ -40,10 +42,9 @@ import (
 // the desired state derived from the given ObjectStores. On creation,
 // the Cluster is set as the owner of the Role for garbage collection.
 //
-// This function is called from both the Pre hook (gRPC) and the
+// This function is called from the Pre hook (gRPC). It creates the
-// ObjectStore controller. To handle concurrent modifications
+// Role if it does not exist, then patches rules and labels to match
-// gracefully, AlreadyExists on Create and Conflict on Patch are
+// the desired state.
-// retried once rather than returned as errors.
 func EnsureRole(
 	ctx context.Context,
 	c client.Client,
@@ -51,108 +52,126 @@ func EnsureRole(
 	barmanObjects []barmancloudv1.ObjectStore,
 ) error {
 	newRole := specs.BuildRole(cluster, barmanObjects)
+	roleKey := client.ObjectKeyFromObject(newRole)
 
-	roleKey := client.ObjectKey{
+	if err := ensureRoleExists(ctx, c, cluster, newRole); err != nil {
-		Namespace: newRole.Namespace,
-		Name:      newRole.Name,
-	}
-
-	var role rbacv1.Role
-	err := c.Get(ctx, roleKey, &role)
-
-	switch {
-	case apierrs.IsNotFound(err):
-		role, err := createRole(ctx, c, cluster, newRole)
-		if err != nil {
 		return err
 	}
-		if role == nil {
+
-			// Created successfully, nothing else to do.
+	return patchRole(ctx, c, roleKey, newRole.Rules, map[string]string{
+		metadata.ClusterLabelName: cluster.Name,
+	})
+}
+
+// EnsureRoleRules updates the rules of an existing Role to match
+// the desired state derived from the given ObjectStores. Unlike
+// EnsureRole, this function does not create Roles or set owner
+// references — it only patches rules on Roles that already exist.
+// It is intended for the ObjectStore controller path where no
+// Cluster object is available. Returns nil if the Role does not
+// exist (the Pre hook has not created it yet).
+func EnsureRoleRules(
+	ctx context.Context,
+	c client.Client,
+	roleKey client.ObjectKey,
+	barmanObjects []barmancloudv1.ObjectStore,
+) error {
+	err := patchRole(ctx, c, roleKey, specs.BuildRoleRules(barmanObjects), nil)
+	if apierrs.IsNotFound(err) {
+		log.FromContext(ctx).Debug("Role not found, skipping rule update",
+			"name", roleKey.Name, "namespace", roleKey.Namespace)
 		return nil
 	}
-		// AlreadyExists: fall through to patch with the re-read role.
-		return patchRoleRules(ctx, c, newRole.Rules, role)
 
-	case err != nil:
 	return err
-
-	default:
-		return patchRoleRules(ctx, c, newRole.Rules, &role)
-	}
 }
 
-// createRole attempts to create the Role. If another writer created
+// ensureRoleExists creates the Role if it does not exist. Returns
-// it concurrently (AlreadyExists), it re-reads and returns the
+// nil on success and nil on AlreadyExists (another writer created
-// existing Role for the caller to patch. On success it returns nil.
+// it concurrently). The caller always follows up with patchRole.
-func createRole(
+func ensureRoleExists(
 	ctx context.Context,
 	c client.Client,
 	cluster *cnpgv1.Cluster,
 	newRole *rbacv1.Role,
-) (*rbacv1.Role, error) {
+) error {
 	contextLogger := log.FromContext(ctx)
 
+	var existing rbacv1.Role
+	err := c.Get(ctx, client.ObjectKeyFromObject(newRole), &existing)
+	if err == nil {
+		return nil
+	}
+	if !apierrs.IsNotFound(err) {
+		return err
+	}
+
 	if err := controllerutil.SetControllerReference(cluster, newRole, c.Scheme()); err != nil {
-		return nil, err
+		return err
 	}
 
 	contextLogger.Info("Creating role",
 		"name", newRole.Name, "namespace", newRole.Namespace)
 
 	createErr := c.Create(ctx, newRole)
-	if createErr == nil {
+	if createErr == nil || apierrs.IsAlreadyExists(createErr) {
-		return nil, nil
+		return nil
-	}
-	if !apierrs.IsAlreadyExists(createErr) {
-		return nil, createErr
 	}
 
-	contextLogger.Info("Role was created concurrently, checking rules")
+	return createErr
-
-	var role rbacv1.Role
-	if err := c.Get(ctx, client.ObjectKeyFromObject(newRole), &role); err != nil {
-		return nil, err
 }
 
-	return &role, nil
+// patchRole patches the Role's rules and optionally its labels to
-}
+// match the desired state. When desiredLabels is nil, labels are
-
+// not modified. Uses retry.RetryOnConflict for concurrent
-// patchRoleRules patches the Role's rules if they differ from the
+// modification handling.
-// desired state. On Conflict (concurrent modification), it retries
+func patchRole(
-// once with a fresh read.
-func patchRoleRules(
 	ctx context.Context,
 	c client.Client,
+	roleKey client.ObjectKey,
 	desiredRules []rbacv1.PolicyRule,
-	role *rbacv1.Role,
+	desiredLabels map[string]string,
 ) error {
-	if equality.Semantic.DeepEqual(desiredRules, role.Rules) {
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		var role rbacv1.Role
+		if err := c.Get(ctx, roleKey, &role); err != nil {
+			return err
+		}
+
+		rulesMatch := equality.Semantic.DeepEqual(desiredRules, role.Rules)
+		labelsMatch := desiredLabels == nil || !labelsNeedUpdate(role.Labels, desiredLabels)
+
+		if rulesMatch && labelsMatch {
 			return nil
 		}
 
 		contextLogger := log.FromContext(ctx)
 		contextLogger.Info("Patching role",
-		"name", role.Name, "namespace", role.Namespace, "rules", desiredRules)
+			"name", role.Name, "namespace", role.Namespace)
 
 		oldRole := role.DeepCopy()
 		role.Rules = desiredRules
 
-	patchErr := c.Patch(ctx, role, client.MergeFrom(oldRole))
+		if desiredLabels != nil {
-	if patchErr == nil || !apierrs.IsConflict(patchErr) {
+			if role.Labels == nil {
-		return patchErr
+				role.Labels = make(map[string]string, len(desiredLabels))
+			}
+			for k, v := range desiredLabels {
+				role.Labels[k] = v
+			}
 		}
 
-	// Conflict: re-read and retry once.
+		return c.Patch(ctx, &role, client.MergeFrom(oldRole))
-	contextLogger.Info("Role was modified concurrently, retrying patch")
+	})
-	if err := c.Get(ctx, client.ObjectKeyFromObject(role), role); err != nil {
-		return err
-	}
-	if equality.Semantic.DeepEqual(desiredRules, role.Rules) {
-		return nil
 }
 
-	oldRole = role.DeepCopy()
+// labelsNeedUpdate returns true if any key in desired is missing
-	role.Rules = desiredRules
+// or has a different value in existing.
-
+func labelsNeedUpdate(existing, desired map[string]string) bool {
-	return c.Patch(ctx, role, client.MergeFrom(oldRole))
+	for k, v := range desired {
+		if existing[k] != v {
+			return true
+		}
+	}
+	return false
 }

internal/cnpgi/operator/rbac/ensure_test.go

@@ -30,18 +30,20 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/rbac"
 )
 
 func newScheme() *runtime.Scheme {
 	s := runtime.NewScheme()
-	_ = rbacv1.AddToScheme(s)
+	utilruntime.Must(rbacv1.AddToScheme(s))
-	_ = cnpgv1.AddToScheme(s)
+	utilruntime.Must(cnpgv1.AddToScheme(s))
-	_ = barmancloudv1.AddToScheme(s)
+	utilruntime.Must(barmancloudv1.AddToScheme(s))
 	return s
 }
 
@@ -99,7 +101,7 @@ var _ = Describe("EnsureRole", func() {
 			fakeClient = fake.NewClientBuilder().WithScheme(newScheme()).Build()
 		})
 
-		It("should create the Role with owner reference", func() {
+		It("should create the Role with owner reference and label", func() {
 			err := rbac.EnsureRole(ctx, fakeClient, cluster, objects)
 			Expect(err).NotTo(HaveOccurred())
 
@@ -111,10 +113,11 @@ var _ = Describe("EnsureRole", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(role.Rules).To(HaveLen(3))
 
-			// Verify owner reference is set to the Cluster
 			Expect(role.OwnerReferences).To(HaveLen(1))
 			Expect(role.OwnerReferences[0].Name).To(Equal("test-cluster"))
 			Expect(role.OwnerReferences[0].Kind).To(Equal("Cluster"))
+
+			Expect(role.Labels).To(HaveKeyWithValue(metadata.ClusterLabelName, "test-cluster"))
 		})
 	})
 
@@ -167,9 +170,133 @@ var _ = Describe("EnsureRole", func() {
 			Expect(secretsRule.ResourceNames).To(ContainElement("aws-creds"))
 			Expect(secretsRule.ResourceNames).NotTo(ContainElement("old-secret"))
 
-			// Owner reference must survive the patch
 			Expect(role.OwnerReferences).To(HaveLen(1))
 			Expect(role.OwnerReferences[0].Name).To(Equal("test-cluster"))
 		})
 	})
+
+	Context("when the Role exists without the cluster label (upgrade scenario)", func() {
+		BeforeEach(func() {
+			fakeClient = fake.NewClientBuilder().WithScheme(newScheme()).Build()
+
+			// Create a Role without the label (simulates pre-upgrade state)
+			unlabeledRole := &rbacv1.Role{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-cluster-barman-cloud",
+					Namespace: "default",
+				},
+				Rules: []rbacv1.PolicyRule{},
+			}
+			Expect(fakeClient.Create(ctx, unlabeledRole)).To(Succeed())
+		})
+
+		It("should add the label and update rules", func() {
+			err := rbac.EnsureRole(ctx, fakeClient, cluster, objects)
+			Expect(err).NotTo(HaveOccurred())
+
+			var role rbacv1.Role
+			Expect(fakeClient.Get(ctx, client.ObjectKey{
+				Namespace: "default",
+				Name:      "test-cluster-barman-cloud",
+			}, &role)).To(Succeed())
+
+			Expect(role.Labels).To(HaveKeyWithValue(metadata.ClusterLabelName, "test-cluster"))
+			Expect(role.Rules).To(HaveLen(3))
+		})
+	})
+})
+
+var _ = Describe("EnsureRoleRules", func() {
+	var (
+		ctx        context.Context
+		fakeClient client.Client
+		objects    []barmancloudv1.ObjectStore
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		objects = []barmancloudv1.ObjectStore{
+			newObjectStore("my-store", "default", "aws-creds"),
+		}
+	})
+
+	Context("when the Role exists", func() {
+		BeforeEach(func() {
+			fakeClient = fake.NewClientBuilder().WithScheme(newScheme()).Build()
+
+			// Seed a labeled Role with old rules
+			cluster := newCluster("test-cluster", "default")
+			oldObjects := []barmancloudv1.ObjectStore{
+				newObjectStore("my-store", "default", "old-secret"),
+			}
+			Expect(rbac.EnsureRole(ctx, fakeClient, cluster, oldObjects)).To(Succeed())
+		})
+
+		It("should patch the rules", func() {
+			roleKey := client.ObjectKey{
+				Namespace: "default",
+				Name:      "test-cluster-barman-cloud",
+			}
+			err := rbac.EnsureRoleRules(ctx, fakeClient, roleKey, objects)
+			Expect(err).NotTo(HaveOccurred())
+
+			var role rbacv1.Role
+			Expect(fakeClient.Get(ctx, roleKey, &role)).To(Succeed())
+
+			secretsRule := role.Rules[2]
+			Expect(secretsRule.ResourceNames).To(ContainElement("aws-creds"))
+			Expect(secretsRule.ResourceNames).NotTo(ContainElement("old-secret"))
+		})
+
+		It("should not patch when rules already match", func() {
+			// Seed with the same objects so rules match
+			cluster := newCluster("test-cluster", "default")
+			Expect(rbac.EnsureRole(ctx, fakeClient, cluster, objects)).To(Succeed())
+
+			roleKey := client.ObjectKey{
+				Namespace: "default",
+				Name:      "test-cluster-barman-cloud",
+			}
+
+			var before rbacv1.Role
+			Expect(fakeClient.Get(ctx, roleKey, &before)).To(Succeed())
+
+			Expect(rbac.EnsureRoleRules(ctx, fakeClient, roleKey, objects)).To(Succeed())
+
+			var after rbacv1.Role
+			Expect(fakeClient.Get(ctx, roleKey, &after)).To(Succeed())
+			Expect(after.ResourceVersion).To(Equal(before.ResourceVersion))
+		})
+
+		It("should not modify labels", func() {
+			roleKey := client.ObjectKey{
+				Namespace: "default",
+				Name:      "test-cluster-barman-cloud",
+			}
+
+			var before rbacv1.Role
+			Expect(fakeClient.Get(ctx, roleKey, &before)).To(Succeed())
+
+			Expect(rbac.EnsureRoleRules(ctx, fakeClient, roleKey, objects)).To(Succeed())
+
+			var after rbacv1.Role
+			Expect(fakeClient.Get(ctx, roleKey, &after)).To(Succeed())
+			Expect(after.Labels).To(Equal(before.Labels))
+		})
+	})
+
+	Context("when the Role does not exist", func() {
+		BeforeEach(func() {
+			fakeClient = fake.NewClientBuilder().WithScheme(newScheme()).Build()
+		})
+
+		It("should return nil", func() {
+			roleKey := client.ObjectKey{
+				Namespace: "default",
+				Name:      "nonexistent-barman-cloud",
+			}
+			err := rbac.EnsureRoleRules(ctx, fakeClient, roleKey, objects)
+			Expect(err).NotTo(HaveOccurred())
+		})
+	})
 })

internal/cnpgi/operator/specs/role.go

@@ -21,6 +21,7 @@ package specs
 
 import (
 	"fmt"
+	"slices"
 
 	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/machinery/pkg/stringset"
@@ -28,6 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 )
 
 // BuildRole builds the Role object for this cluster
@@ -35,15 +37,20 @@ func BuildRole(
 	cluster *cnpgv1.Cluster,
 	barmanObjects []barmancloudv1.ObjectStore,
 ) *rbacv1.Role {
-	role := &rbacv1.Role{
+	return &rbacv1.Role{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: cluster.Namespace,
 			Name:      GetRBACName(cluster.Name),
+			Labels: map[string]string{
+				metadata.ClusterLabelName: cluster.Name,
 			},
-
+		},
-		Rules: []rbacv1.PolicyRule{},
+		Rules: BuildRoleRules(barmanObjects),
+	}
 }
 
+// BuildRoleRules builds the RBAC PolicyRules for the given ObjectStores.
+func BuildRoleRules(barmanObjects []barmancloudv1.ObjectStore) []rbacv1.PolicyRule {
 	secretsSet := stringset.New()
 	barmanObjectsSet := stringset.New()
 
@@ -54,11 +61,10 @@ func BuildRole(
 		}
 	}
 
-	role.Rules = append(
+	return []rbacv1.PolicyRule{
-		role.Rules,
+		{
-		rbacv1.PolicyRule{
 			APIGroups: []string{
-				"barmancloud.cnpg.io",
+				barmancloudv1.GroupVersion.Group,
 			},
 			Verbs: []string{
 				"get",
@@ -70,9 +76,9 @@ func BuildRole(
 			},
 			ResourceNames: barmanObjectsSet.ToSortedList(),
 		},
-		rbacv1.PolicyRule{
+		{
 			APIGroups: []string{
-				"barmancloud.cnpg.io",
+				barmancloudv1.GroupVersion.Group,
 			},
 			Verbs: []string{
 				"update",
@@ -82,7 +88,7 @@ func BuildRole(
 			},
 			ResourceNames: barmanObjectsSet.ToSortedList(),
 		},
-		rbacv1.PolicyRule{
+		{
 			APIGroups: []string{
 				"",
 			},
@@ -96,9 +102,25 @@ func BuildRole(
 			},
 			ResourceNames: secretsSet.ToSortedList(),
 		},
-	)
+	}
+}
 
-	return role
+// ObjectStoreNamesFromRole extracts the ObjectStore names referenced
+// by a plugin-managed Role. It finds the objectstores rule
+// semantically (by APIGroup and Resource, not by index) and returns
+// a copy of its ResourceNames. Returns nil if no matching rule is
+// found.
+func ObjectStoreNamesFromRole(role *rbacv1.Role) []string {
+	for _, rule := range role.Rules {
+		if len(rule.APIGroups) == 1 &&
+			rule.APIGroups[0] == barmancloudv1.GroupVersion.Group &&
+			len(rule.Resources) == 1 &&
+			rule.Resources[0] == "objectstores" {
+			return slices.Clone(rule.ResourceNames)
+		}
+	}
+
+	return nil
 }
 
 // BuildRoleBinding builds the role binding object for this cluster

internal/cnpgi/operator/specs/role_test.go

@@ -0,0 +1,210 @@
+/*
+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 specs
+
+import (
+	barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
+	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	machineryapi "github.com/cloudnative-pg/machinery/pkg/api"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
+)
+
+func newTestObjectStore(name, secretName string) barmancloudv1.ObjectStore {
+	return barmancloudv1.ObjectStore{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "default",
+		},
+		Spec: barmancloudv1.ObjectStoreSpec{
+			Configuration: barmanapi.BarmanObjectStoreConfiguration{
+				DestinationPath: "s3://bucket/path",
+				BarmanCredentials: barmanapi.BarmanCredentials{
+					AWS: &barmanapi.S3Credentials{
+						AccessKeyIDReference: &machineryapi.SecretKeySelector{
+							LocalObjectReference: machineryapi.LocalObjectReference{
+								Name: secretName,
+							},
+							Key: "ACCESS_KEY_ID",
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+var _ = Describe("BuildRoleRules", func() {
+	It("should produce 3 rules with correct ResourceNames", func() {
+		objects := []barmancloudv1.ObjectStore{
+			newTestObjectStore("store-a", "secret-a"),
+			newTestObjectStore("store-b", "secret-b"),
+		}
+		rules := BuildRoleRules(objects)
+		Expect(rules).To(HaveLen(3))
+
+		Expect(rules[0].APIGroups).To(Equal([]string{barmancloudv1.GroupVersion.Group}))
+		Expect(rules[0].Resources).To(Equal([]string{"objectstores"}))
+		Expect(rules[0].ResourceNames).To(ConsistOf("store-a", "store-b"))
+
+		Expect(rules[1].APIGroups).To(Equal([]string{barmancloudv1.GroupVersion.Group}))
+		Expect(rules[1].Resources).To(Equal([]string{"objectstores/status"}))
+		Expect(rules[1].ResourceNames).To(ConsistOf("store-a", "store-b"))
+
+		Expect(rules[2].APIGroups).To(Equal([]string{""}))
+		Expect(rules[2].Resources).To(Equal([]string{"secrets"}))
+		Expect(rules[2].ResourceNames).To(ConsistOf("secret-a", "secret-b"))
+	})
+
+	It("should produce rules with empty ResourceNames for empty input", func() {
+		rules := BuildRoleRules(nil)
+		Expect(rules).To(HaveLen(3))
+		Expect(rules[0].ResourceNames).To(BeEmpty())
+		Expect(rules[0].ResourceNames).NotTo(BeNil())
+		Expect(rules[1].ResourceNames).To(BeEmpty())
+		Expect(rules[2].ResourceNames).To(BeEmpty())
+	})
+
+	It("should deduplicate secret names across ObjectStores", func() {
+		objects := []barmancloudv1.ObjectStore{
+			newTestObjectStore("store-a", "shared-secret"),
+			newTestObjectStore("store-b", "shared-secret"),
+		}
+		rules := BuildRoleRules(objects)
+		Expect(rules[2].ResourceNames).To(Equal([]string{"shared-secret"}))
+	})
+})
+
+var _ = Describe("BuildRole", func() {
+	It("should set the cluster label", func() {
+		cluster := &cnpgv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "my-cluster",
+				Namespace: "default",
+			},
+		}
+		role := BuildRole(cluster, nil)
+		Expect(role.Labels).To(HaveKeyWithValue(metadata.ClusterLabelName, "my-cluster"))
+		Expect(role.Name).To(Equal("my-cluster-barman-cloud"))
+		Expect(role.Namespace).To(Equal("default"))
+	})
+})
+
+var _ = Describe("BuildRoleRules / ObjectStoreNamesFromRole round-trip", func() {
+	It("should recover the same ObjectStore names from built rules", func() {
+		objects := []barmancloudv1.ObjectStore{
+			newTestObjectStore("store-a", "secret-a"),
+			newTestObjectStore("store-b", "secret-b"),
+		}
+		rules := BuildRoleRules(objects)
+		role := &rbacv1.Role{Rules: rules}
+		names := ObjectStoreNamesFromRole(role)
+		Expect(names).To(ConsistOf("store-a", "store-b"))
+	})
+
+	It("should recover empty names from rules built with no ObjectStores", func() {
+		rules := BuildRoleRules(nil)
+		role := &rbacv1.Role{Rules: rules}
+		names := ObjectStoreNamesFromRole(role)
+		Expect(names).To(BeEmpty())
+	})
+})
+
+var _ = Describe("ObjectStoreNamesFromRole", func() {
+	It("should extract ObjectStore names from a well-formed Role", func() {
+		role := &rbacv1.Role{
+			Rules: []rbacv1.PolicyRule{
+				{
+					APIGroups:     []string{barmancloudv1.GroupVersion.Group},
+					Resources:     []string{"objectstores"},
+					ResourceNames: []string{"store-a", "store-b"},
+				},
+				{
+					APIGroups:     []string{""},
+					Resources:     []string{"secrets"},
+					ResourceNames: []string{"secret-a"},
+				},
+			},
+		}
+		Expect(ObjectStoreNamesFromRole(role)).To(Equal([]string{"store-a", "store-b"}))
+	})
+
+	It("should return nil for a Role with no matching rule", func() {
+		role := &rbacv1.Role{
+			Rules: []rbacv1.PolicyRule{
+				{
+					APIGroups:     []string{""},
+					Resources:     []string{"secrets"},
+					ResourceNames: []string{"secret-a"},
+				},
+			},
+		}
+		Expect(ObjectStoreNamesFromRole(role)).To(BeNil())
+	})
+
+	It("should return nil for a Role with empty rules", func() {
+		role := &rbacv1.Role{}
+		Expect(ObjectStoreNamesFromRole(role)).To(BeNil())
+	})
+
+	It("should not match a rule with a different APIGroup", func() {
+		role := &rbacv1.Role{
+			Rules: []rbacv1.PolicyRule{
+				{
+					APIGroups:     []string{"other.io"},
+					Resources:     []string{"objectstores"},
+					ResourceNames: []string{"store-a"},
+				},
+			},
+		}
+		Expect(ObjectStoreNamesFromRole(role)).To(BeNil())
+	})
+
+	It("should not match a rule with multiple APIGroups", func() {
+		role := &rbacv1.Role{
+			Rules: []rbacv1.PolicyRule{
+				{
+					APIGroups:     []string{barmancloudv1.GroupVersion.Group, "other.io"},
+					Resources:     []string{"objectstores"},
+					ResourceNames: []string{"store-a"},
+				},
+			},
+		}
+		Expect(ObjectStoreNamesFromRole(role)).To(BeNil())
+	})
+
+	It("should not match a rule for objectstores/status", func() {
+		role := &rbacv1.Role{
+			Rules: []rbacv1.PolicyRule{
+				{
+					APIGroups:     []string{barmancloudv1.GroupVersion.Group},
+					Resources:     []string{"objectstores/status"},
+					ResourceNames: []string{"store-a"},
+				},
+			},
+		}
+		Expect(ObjectStoreNamesFromRole(role)).To(BeNil())
+	})
+})

internal/controller/objectstore_controller.go

@@ -21,10 +21,12 @@ package controller
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"slices"
 
-	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/machinery/pkg/log"
+	rbacv1 "k8s.io/api/rbac/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -33,8 +35,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
-	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/config"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/rbac"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/specs"
 )
 
 // ObjectStoreReconciler reconciles a ObjectStore object.
@@ -46,15 +49,16 @@ type ObjectStoreReconciler struct {
 // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebindings,verbs=create;patch;update;get;list;watch
 // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=roles,verbs=create;patch;update;get;list;watch
 // +kubebuilder:rbac:groups="",resources=secrets,verbs=create;list;get;watch;delete
-// +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=clusters,verbs=get;list;watch
-// +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=clusters/finalizers,verbs=update
 // +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=backups,verbs=get;list;watch
+// +kubebuilder:rbac:groups=postgresql.cnpg.io,resources=clusters/finalizers,verbs=update
 // +kubebuilder:rbac:groups=barmancloud.cnpg.io,resources=objectstores,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=barmancloud.cnpg.io,resources=objectstores/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=barmancloud.cnpg.io,resources=objectstores/finalizers,verbs=update
 
 // Reconcile ensures that the RBAC Role for each Cluster referencing
 // this ObjectStore is up to date with the current ObjectStore spec.
+// It discovers affected Roles by listing plugin-managed Roles and
+// inspecting their rules, without needing access to Cluster objects.
 func (r *ObjectStoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	contextLogger := log.FromContext(ctx).WithValues(
 		"objectStoreName", req.Name,
@@ -64,72 +68,64 @@ func (r *ObjectStoreReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 
 	contextLogger.Info("ObjectStore reconciliation start")
 
-	// List all Clusters in the same namespace
+	var roleList rbacv1.RoleList
-	var clusterList cnpgv1.ClusterList
+	if err := r.List(ctx, &roleList,
-	if err := r.List(ctx, &clusterList, client.InNamespace(req.Namespace)); err != nil {
+		client.InNamespace(req.Namespace),
-		return ctrl.Result{}, fmt.Errorf("while listing clusters: %w", err)
+		client.HasLabels{metadata.ClusterLabelName},
+	); err != nil {
+		return ctrl.Result{}, fmt.Errorf("while listing roles: %w", err)
 	}
 
-	// For each Cluster that references this ObjectStore, reconcile the Role
+	var errs []error
-	for i := range clusterList.Items {
+	for i := range roleList.Items {
-		cluster := &clusterList.Items[i]
+		role := &roleList.Items[i]
 
-		pluginConfiguration := config.NewFromCluster(cluster)
+		objectStoreNames := specs.ObjectStoreNamesFromRole(role)
-		referredObjects := pluginConfiguration.GetReferredBarmanObjectsKey()
+		if !slices.Contains(objectStoreNames, req.Name) {
-
-		if !referencesObjectStore(referredObjects, req.NamespacedName) {
 			continue
 		}
 
-		contextLogger.Info("Reconciling RBAC for cluster",
+		contextLogger.Info("Reconciling RBAC for role",
-			"clusterName", cluster.Name)
+			"roleName", role.Name)
 
-		if err := r.reconcileRBACForCluster(ctx, cluster, referredObjects); err != nil {
+		if err := r.reconcileRoleRules(ctx, role, objectStoreNames); err != nil {
-			return ctrl.Result{}, fmt.Errorf("while reconciling RBAC for cluster %s: %w", cluster.Name, err)
+			contextLogger.Error(err, "Failed to reconcile RBAC for role",
+				"roleName", role.Name)
+			errs = append(errs, fmt.Errorf("while reconciling role %s: %w", role.Name, err))
 		}
 	}
 
 	contextLogger.Info("ObjectStore reconciliation completed")
-	return ctrl.Result{}, nil
+	return ctrl.Result{}, errors.Join(errs...)
 }
 
-// reconcileRBACForCluster ensures the Role for the given Cluster is
+// reconcileRoleRules fetches the ObjectStores referenced by the
-// up to date with the current ObjectStore specs.
+// Role and patches its rules to match the current specs.
-func (r *ObjectStoreReconciler) reconcileRBACForCluster(
+func (r *ObjectStoreReconciler) reconcileRoleRules(
 	ctx context.Context,
-	cluster *cnpgv1.Cluster,
+	role *rbacv1.Role,
-	referredObjectKeys []client.ObjectKey,
+	objectStoreNames []string,
 ) error {
 	contextLogger := log.FromContext(ctx)
-	barmanObjects := make([]barmancloudv1.ObjectStore, 0, len(referredObjectKeys))
+	barmanObjects := make([]barmancloudv1.ObjectStore, 0, len(objectStoreNames))
-	for _, key := range referredObjectKeys {
+
+	for _, name := range objectStoreNames {
 		var barmanObject barmancloudv1.ObjectStore
-		if err := r.Get(ctx, key, &barmanObject); err != nil {
+		if err := r.Get(ctx, client.ObjectKey{
+			Namespace: role.Namespace,
+			Name:      name,
+		}, &barmanObject); err != nil {
 			if apierrs.IsNotFound(err) {
 				contextLogger.Info("ObjectStore not found, skipping",
-					"objectStoreName", key.Name)
+					"objectStoreName", name)
 				continue
 			}
-			return fmt.Errorf("while getting ObjectStore %s: %w", key, err)
+			return fmt.Errorf("while getting ObjectStore %s: %w", name, err)
 		}
 		barmanObjects = append(barmanObjects, barmanObject)
 	}
 
-	return rbac.EnsureRole(ctx, r.Client, cluster, barmanObjects)
+	return rbac.EnsureRoleRules(ctx, r.Client, client.ObjectKeyFromObject(role), barmanObjects)
-}
-
-// referencesObjectStore checks if the given ObjectStore is in the list
-// of referred barman objects.
-func referencesObjectStore(
-	referredObjects []client.ObjectKey,
-	objectStore client.ObjectKey,
-) bool {
-	for _, ref := range referredObjects {
-		if ref.Name == objectStore.Name && ref.Namespace == objectStore.Namespace {
-			return true
-		}
-	}
-	return false
 }
 
 // SetupWithManager sets up the controller with the Manager.

internal/controller/objectstore_controller_test.go

@@ -22,7 +22,6 @@ package controller
 import (
 	"context"
 
-	cnpgv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	barmanapi "github.com/cloudnative-pg/barman-cloud/pkg/api"
 	machineryapi "github.com/cloudnative-pg/machinery/pkg/api"
 	. "github.com/onsi/ginkgo/v2"
@@ -30,6 +29,7 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -37,35 +37,16 @@ import (
 
 	barmancloudv1 "github.com/cloudnative-pg/plugin-barman-cloud/api/v1"
 	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/metadata"
+	"github.com/cloudnative-pg/plugin-barman-cloud/internal/cnpgi/operator/specs"
 )
 
 func newFakeScheme() *runtime.Scheme {
 	s := runtime.NewScheme()
-	_ = rbacv1.AddToScheme(s)
+	utilruntime.Must(rbacv1.AddToScheme(s))
-	_ = cnpgv1.AddToScheme(s)
+	utilruntime.Must(barmancloudv1.AddToScheme(s))
-	_ = barmancloudv1.AddToScheme(s)
 	return s
 }
 
-func newTestCluster(name, namespace, objectStoreName string) *cnpgv1.Cluster {
-	return &cnpgv1.Cluster{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-		Spec: cnpgv1.ClusterSpec{
-			Plugins: []cnpgv1.PluginConfiguration{
-				{
-					Name: metadata.PluginName,
-					Parameters: map[string]string{
-						"barmanObjectName": objectStoreName,
-					},
-				},
-			},
-		},
-	}
-}
-
 func newTestObjectStore(name, namespace, secretName string) *barmancloudv1.ObjectStore {
 	return &barmancloudv1.ObjectStore{
 		ObjectMeta: metav1.ObjectMeta{
@@ -90,41 +71,18 @@ func newTestObjectStore(name, namespace, secretName string) *barmancloudv1.Objec
 	}
 }
 
-var _ = Describe("referencesObjectStore", func() {
+func newLabeledRole(clusterName, namespace string, objectStores []barmancloudv1.ObjectStore) *rbacv1.Role {
-	It("should return true when ObjectStore is in the list", func() {
+	return &rbacv1.Role{
-		refs := []client.ObjectKey{
+		ObjectMeta: metav1.ObjectMeta{
-			{Name: "store-a", Namespace: "default"},
+			Name:      specs.GetRBACName(clusterName),
-			{Name: "store-b", Namespace: "default"},
+			Namespace: namespace,
+			Labels: map[string]string{
+				metadata.ClusterLabelName: clusterName,
+			},
+		},
+		Rules: specs.BuildRoleRules(objectStores),
 	}
-		Expect(referencesObjectStore(refs, client.ObjectKey{
-			Name: "store-b", Namespace: "default",
-		})).To(BeTrue())
-	})
-
-	It("should return false when ObjectStore is not in the list", func() {
-		refs := []client.ObjectKey{
-			{Name: "store-a", Namespace: "default"},
 }
-		Expect(referencesObjectStore(refs, client.ObjectKey{
-			Name: "store-b", Namespace: "default",
-		})).To(BeFalse())
-	})
-
-	It("should return false when namespace differs", func() {
-		refs := []client.ObjectKey{
-			{Name: "store-a", Namespace: "ns1"},
-		}
-		Expect(referencesObjectStore(refs, client.ObjectKey{
-			Name: "store-a", Namespace: "ns2",
-		})).To(BeFalse())
-	})
-
-	It("should return false for empty list", func() {
-		Expect(referencesObjectStore(nil, client.ObjectKey{
-			Name: "store-a", Namespace: "default",
-		})).To(BeFalse())
-	})
-})
 
 var _ = Describe("ObjectStoreReconciler", func() {
 	var (
@@ -138,185 +96,256 @@ var _ = Describe("ObjectStoreReconciler", func() {
 	})
 
 	Describe("Reconcile", func() {
-		It("should create a Role for a Cluster that references the ObjectStore", func() {
+		It("should update Role rules when ObjectStore credentials change", func() {
-			objectStore := newTestObjectStore("my-store", "default", "aws-creds")
-			cluster := newTestCluster("my-cluster", "default", "my-store")
-
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(objectStore, cluster).
-				Build()
-
-			reconciler := &ObjectStoreReconciler{
-				Client: fakeClient,
-				Scheme: scheme,
-			}
-
-			result, err := reconciler.Reconcile(ctx, reconcile.Request{
-				NamespacedName: types.NamespacedName{
-					Name:      "my-store",
-					Namespace: "default",
-				},
-			})
-			Expect(err).NotTo(HaveOccurred())
-			Expect(result).To(Equal(reconcile.Result{}))
-
-			var role rbacv1.Role
-			err = fakeClient.Get(ctx, client.ObjectKey{
-				Namespace: "default",
-				Name:      "my-cluster-barman-cloud",
-			}, &role)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(role.Rules).To(HaveLen(3))
-
-			// Verify the secrets rule contains the expected secret
-			secretsRule := role.Rules[2]
-			Expect(secretsRule.ResourceNames).To(ContainElement("aws-creds"))
-
-			// Verify owner reference is set to the Cluster
-			Expect(role.OwnerReferences).To(HaveLen(1))
-			Expect(role.OwnerReferences[0].Name).To(Equal("my-cluster"))
-			Expect(role.OwnerReferences[0].Kind).To(Equal("Cluster"))
-		})
-
-		It("should skip Clusters that don't reference the ObjectStore", func() {
-			objectStore := newTestObjectStore("my-store", "default", "aws-creds")
-			cluster := newTestCluster("my-cluster", "default", "other-store")
-
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(objectStore, cluster).
-				Build()
-
-			reconciler := &ObjectStoreReconciler{
-				Client: fakeClient,
-				Scheme: scheme,
-			}
-
-			result, err := reconciler.Reconcile(ctx, reconcile.Request{
-				NamespacedName: types.NamespacedName{
-					Name:      "my-store",
-					Namespace: "default",
-				},
-			})
-			Expect(err).NotTo(HaveOccurred())
-			Expect(result).To(Equal(reconcile.Result{}))
-
-			// No Role should have been created
-			var role rbacv1.Role
-			err = fakeClient.Get(ctx, client.ObjectKey{
-				Namespace: "default",
-				Name:      "my-cluster-barman-cloud",
-			}, &role)
-			Expect(err).To(HaveOccurred())
-		})
-
-		It("should succeed with no Clusters in the namespace", func() {
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				Build()
-
-			reconciler := &ObjectStoreReconciler{
-				Client: fakeClient,
-				Scheme: scheme,
-			}
-
-			result, err := reconciler.Reconcile(ctx, reconcile.Request{
-				NamespacedName: types.NamespacedName{
-					Name:      "my-store",
-					Namespace: "default",
-				},
-			})
-			Expect(err).NotTo(HaveOccurred())
-			Expect(result).To(Equal(reconcile.Result{}))
-		})
-	})
-
-	Describe("reconcileRBACForCluster", func() {
-		It("should skip deleted ObjectStores and still reconcile the Role", func() {
-			// Cluster references two ObjectStores, but one is deleted
-			cluster := newTestCluster("my-cluster", "default", "store-a")
-			existingStore := newTestObjectStore("store-a", "default", "aws-creds")
-
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(existingStore).
-				Build()
-
-			reconciler := &ObjectStoreReconciler{
-				Client: fakeClient,
-				Scheme: scheme,
-			}
-
-			// Pass two keys, but "store-b" doesn't exist
-			err := reconciler.reconcileRBACForCluster(ctx, cluster, []client.ObjectKey{
-				{Name: "store-a", Namespace: "default"},
-				{Name: "store-b", Namespace: "default"},
-			})
-			Expect(err).NotTo(HaveOccurred())
-
-			// Role should be created with only store-a's secrets
-			var role rbacv1.Role
-			err = fakeClient.Get(ctx, client.ObjectKey{
-				Namespace: "default",
-				Name:      "my-cluster-barman-cloud",
-			}, &role)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(role.Rules).To(HaveLen(3))
-
-			// ObjectStore rule should only reference store-a
-			objectStoreRule := role.Rules[0]
-			Expect(objectStoreRule.ResourceNames).To(ContainElement("store-a"))
-			Expect(objectStoreRule.ResourceNames).NotTo(ContainElement("store-b"))
-
-			// Verify owner reference is set
-			Expect(role.OwnerReferences).To(HaveLen(1))
-			Expect(role.OwnerReferences[0].Name).To(Equal("my-cluster"))
-		})
-
-		It("should update Role when ObjectStore credentials change", func() {
-			cluster := newTestCluster("my-cluster", "default", "my-store")
 			oldStore := newTestObjectStore("my-store", "default", "old-secret")
-
+			role := newLabeledRole("my-cluster", "default", []barmancloudv1.ObjectStore{*oldStore})
-			fakeClient := fake.NewClientBuilder().
-				WithScheme(scheme).
-				WithObjects(oldStore).
-				Build()
-
-			reconciler := &ObjectStoreReconciler{
-				Client: fakeClient,
-				Scheme: scheme,
-			}
-
-			// First reconcile - creates Role with old-secret
-			err := reconciler.reconcileRBACForCluster(ctx, cluster, []client.ObjectKey{
-				{Name: "my-store", Namespace: "default"},
-			})
-			Expect(err).NotTo(HaveOccurred())
 
 			// Update the ObjectStore with new credentials
-			var currentStore barmancloudv1.ObjectStore
+			newStore := newTestObjectStore("my-store", "default", "new-secret")
-			Expect(fakeClient.Get(ctx, client.ObjectKey{
-				Name: "my-store", Namespace: "default",
-			}, &currentStore)).To(Succeed())
-			currentStore.Spec.Configuration.BarmanCredentials.AWS.AccessKeyIDReference.LocalObjectReference.Name = "new-secret"
-			Expect(fakeClient.Update(ctx, &currentStore)).To(Succeed())
 
-			// Second reconcile - should patch Role with new-secret
+			fakeClient := fake.NewClientBuilder().
-			err = reconciler.reconcileRBACForCluster(ctx, cluster, []client.ObjectKey{
+				WithScheme(scheme).
-				{Name: "my-store", Namespace: "default"},
+				WithObjects(role, newStore).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "my-store",
+					Namespace: "default",
+				},
 			})
 			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
 
-			var role rbacv1.Role
+			var updatedRole rbacv1.Role
 			Expect(fakeClient.Get(ctx, client.ObjectKey{
 				Namespace: "default",
 				Name:      "my-cluster-barman-cloud",
-			}, &role)).To(Succeed())
+			}, &updatedRole)).To(Succeed())
 
-			secretsRule := role.Rules[2]
+			secretsRule := updatedRole.Rules[2]
 			Expect(secretsRule.ResourceNames).To(ContainElement("new-secret"))
 			Expect(secretsRule.ResourceNames).NotTo(ContainElement("old-secret"))
 		})
+
+		It("should skip Roles that don't reference the ObjectStore", func() {
+			otherStore := newTestObjectStore("other-store", "default", "other-creds")
+			role := newLabeledRole("my-cluster", "default", []barmancloudv1.ObjectStore{*otherStore})
+
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(role).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			var before rbacv1.Role
+			Expect(fakeClient.Get(ctx, client.ObjectKey{
+				Namespace: "default",
+				Name:      "my-cluster-barman-cloud",
+			}, &before)).To(Succeed())
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "unrelated-store",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+
+			var after rbacv1.Role
+			Expect(fakeClient.Get(ctx, client.ObjectKey{
+				Namespace: "default",
+				Name:      "my-cluster-barman-cloud",
+			}, &after)).To(Succeed())
+
+			Expect(after.ResourceVersion).To(Equal(before.ResourceVersion))
+		})
+
+		It("should succeed with no labeled Roles in the namespace", func() {
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "my-store",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+		})
+
+		It("should handle deleted ObjectStores gracefully", func() {
+			storeA := newTestObjectStore("store-a", "default", "secret-a")
+			storeB := newTestObjectStore("store-b", "default", "secret-b")
+			role := newLabeledRole("my-cluster", "default", []barmancloudv1.ObjectStore{*storeA, *storeB})
+
+			// Only store-a exists; store-b was deleted
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(role, storeA).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "store-b",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+
+			var updatedRole rbacv1.Role
+			Expect(fakeClient.Get(ctx, client.ObjectKey{
+				Namespace: "default",
+				Name:      "my-cluster-barman-cloud",
+			}, &updatedRole)).To(Succeed())
+
+			objectStoreRule := updatedRole.Rules[0]
+			Expect(objectStoreRule.ResourceNames).To(ContainElement("store-a"))
+			Expect(objectStoreRule.ResourceNames).NotTo(ContainElement("store-b"))
+		})
+
+		It("should not panic on a Role with empty rules", func() {
+			emptyRole := &rbacv1.Role{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "empty-barman-cloud",
+					Namespace: "default",
+					Labels: map[string]string{
+						metadata.ClusterLabelName: "empty",
+					},
+				},
+			}
+
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(emptyRole).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "my-store",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+		})
+
+		It("should produce empty ResourceNames when all ObjectStores are deleted", func() {
+			store := newTestObjectStore("my-store", "default", "aws-creds")
+			role := newLabeledRole("my-cluster", "default", []barmancloudv1.ObjectStore{*store})
+
+			// Don't add the ObjectStore to the fake client (simulates deletion)
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(role).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "my-store",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+
+			var updatedRole rbacv1.Role
+			Expect(fakeClient.Get(ctx, client.ObjectKey{
+				Namespace: "default",
+				Name:      "my-cluster-barman-cloud",
+			}, &updatedRole)).To(Succeed())
+
+			// All rules should have empty ResourceNames
+			Expect(updatedRole.Rules[0].ResourceNames).To(BeEmpty())
+			Expect(updatedRole.Rules[1].ResourceNames).To(BeEmpty())
+			Expect(updatedRole.Rules[2].ResourceNames).To(BeEmpty())
+		})
+
+		It("should reconcile multiple Roles referencing the same ObjectStore", func() {
+			store := newTestObjectStore("shared-store", "default", "new-secret")
+			oldStore := barmancloudv1.ObjectStore{
+				ObjectMeta: metav1.ObjectMeta{Name: "shared-store", Namespace: "default"},
+				Spec: barmancloudv1.ObjectStoreSpec{
+					Configuration: barmanapi.BarmanObjectStoreConfiguration{
+						DestinationPath: "s3://bucket/path",
+						BarmanCredentials: barmanapi.BarmanCredentials{
+							AWS: &barmanapi.S3Credentials{
+								AccessKeyIDReference: &machineryapi.SecretKeySelector{
+									LocalObjectReference: machineryapi.LocalObjectReference{Name: "old-secret"},
+									Key:                  "ACCESS_KEY_ID",
+								},
+							},
+						},
+					},
+				},
+			}
+
+			role1 := newLabeledRole("cluster-1", "default", []barmancloudv1.ObjectStore{oldStore})
+			role2 := newLabeledRole("cluster-2", "default", []barmancloudv1.ObjectStore{oldStore})
+
+			fakeClient := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(role1, role2, store).
+				Build()
+
+			reconciler := &ObjectStoreReconciler{
+				Client: fakeClient,
+				Scheme: scheme,
+			}
+
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      "shared-store",
+					Namespace: "default",
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result).To(Equal(reconcile.Result{}))
+
+			for _, clusterName := range []string{"cluster-1", "cluster-2"} {
+				var updatedRole rbacv1.Role
+				Expect(fakeClient.Get(ctx, client.ObjectKey{
+					Namespace: "default",
+					Name:      specs.GetRBACName(clusterName),
+				}, &updatedRole)).To(Succeed())
+
+				secretsRule := updatedRole.Rules[2]
+				Expect(secretsRule.ResourceNames).To(ContainElement("new-secret"))
+				Expect(secretsRule.ResourceNames).NotTo(ContainElement("old-secret"))
+			}
+		})
 	})
 })

manifest.yaml

@@ -870,7 +870,6 @@ rules:
   - postgresql.cnpg.io
   resources:
   - backups
-  - clusters
   verbs:
   - get
   - list