feat(component): block a resource whose live object is controlled by another owner

docs/component.md

@@ -65,6 +65,7 @@ be passed without a guard.
 | `component.Auxiliary()`                             | The resource's health does not contribute to the component condition (a blocked guard still does)                                                                                                                                       |
 | `component.BlockOnAbsence()`                        | Read-only only: a NotFound records a blocked status and short-circuits the remaining resources                                                                                                                                          |
 | `component.IgnoreIfAbsent()`                        | Read-only only: a NotFound is silently ignored and last-known state is preserved                                                                                                                                                        |
+| `component.BlockOnForeignController()`              | Managed only: records a blocked status that names the owner whose controller reference is on the live object, then skips the apply and the remaining resources                                                                          |
 | `component.SuppressGraceInconsistencyWarning()`     | Suppresses the grace/convergence inconsistency warning                                                                                                                                                                                  |
 
 A read-only resource is not owned by the component, so it is never deleted. `ReadOnly()` is mutually exclusive with
@@ -79,6 +80,31 @@ is still subject to explicit deletion: `Delete()`, `DeleteWhen()`, `GatedBy()` (
 suspension with `DeleteOnSuspend()` all delete it directly, regardless of the `Unowned` flag. Only Kubernetes GC
 (triggered by owner CR deletion) is suppressed.
 
+`BlockOnForeignController()` protects a managed resource from an object that another owner already controls. Before each
+apply, the component reads the live object. If the object has a controller reference to a different owner, the resource
+reports `Blocked` with the message `controlled by <Kind> <name>`. The component performs no apply and skips the
+resources after it, exactly as for a [blocked guard](#guards). The block clears on the first reconcile after that
+reference is gone. An object with no controller reference is never blocked, so the option does not detect two owners
+that both apply without one.
+
+The read goes through `ReconcileContext.APIReader` when it is set, and through `ReconcileContext.Client` otherwise. The
+cached client can miss a controller reference that the API server already has.
+
+Use the option on any resource that two custom resources can name. With the default controller reference, it replaces
+the rejection by the API server of a second controller with a readable condition. With `Unowned()`, it stops the forced
+apply of the second owner from taking the fields of the object (see
+[Server-Side Apply](primitives.md#server-side-apply)).
+
+Unlike a custom guard, the check also covers every path that deletes the object. During suspension, the component does
+not scale down or delete a resource that another owner controls. The resource counts as suspended, so the component
+condition reads `Suspended` with the usual `All resources are suspended.` message. The component also skips a deletion
+that `Delete()`, `DeleteWhen()`, `GatedBy()` or a disabled feature gate asks for. The component logs each skip with the
+controlling owner.
+
+A delete of an object that the read found safe carries the observed UID and resourceVersion as preconditions. If another
+owner claims the object between the read and the delete, the delete fails and the next reconcile reads the object again.
+The option requires a managed resource. A combination with `ReadOnly()` is a build error.
+
 Options compose. Gate a resource and exclude it from health aggregation in one call:
 
 ```go
@@ -1121,8 +1147,13 @@ registered custom guard; it does not affect declared data guards.
   regardless of its participation mode, and all resources after it are skipped entirely. This override exists because a
   blocked guard halts the entire pipeline; subsequent required resources would otherwise be silently absent from health
   aggregation.
+- After the guard of a resource clears, the component also reads the controller reference of the live object for a
+  resource registered with [`BlockOnForeignController()`](#resource-registration-options). A reference to another owner
+  records `Blocked` in the same way, with the message `controlled by <Kind> <name>`.
 - On the next reconcile, if the guard clears (`Unblocked`), the resource is applied normally.
-- Guards are **not** evaluated during suspension. The suspension path always proceeds regardless of guard state.
+- Guards are **not** evaluated during suspension. The suspension path always proceeds regardless of guard state. The
+  exception is [`BlockOnForeignController()`](#resource-registration-options), which the component checks on every path.
+  As a result, a suspension never scales down or deletes an object that another owner controls.
 - A guard evaluation error is treated as a reconciliation failure and sets the condition to `Error`.
 
 A blocked guard produces a condition like:

docs/primitives.md

@@ -149,10 +149,15 @@ converged, and neither would ever see a conflict. Naming the owner also makes
 
 The rejection depends on the controller reference. For a resource registered with `Unowned()`, or one whose owner
 reference cannot be set because of a scope mismatch, nothing stops the second owner's forced apply from taking the
-fields it declares, and the fields move between the two owners' managers on every reconcile. `managedFields` then names
-the owner that wrote each field, but the framework does not detect the contention. A shared name between two owners is
-the operator's responsibility in that case; a [guard](component.md#guards) that reads the live object and blocks when
-another owner controls it is the way to make it explicit.
+fields it declares, and the fields move between the two owners' managers on every reconcile.
+
+Register the resource with [`component.BlockOnForeignController()`](component.md#resource-registration-options) to make
+the contention visible. Before each apply, the component reads the live object. If the object has a controller reference
+to another owner, the resource reports `Blocked` and names that owner instead of applying. In the default case, this
+also turns the rejection by the API server into a readable condition. The check compares controller references only, so
+two owners that both apply without one leave no identity on the object (two `Unowned()` registrations, or owners that
+the scope of the object keeps from being referenced). Then the fields keep moving between the two managers, and a shared
+name remains the responsibility of the operator.
 
 !!! note "Upgrading from a release without the UID in the manager name"
 

pkg/component/builder.go

@@ -114,8 +114,8 @@ func (b *Builder) WithConditionType(conditionType ConditionType) *Builder {
 //
 // Options configure the resource's lifecycle and its participation in health
 // aggregation; see the ResourceOption constructors (ReadOnly, Delete, DeleteWhen,
-// GatedBy, Auxiliary, BlockOnAbsence, IgnoreIfAbsent,
-// SuppressGraceInconsistencyWarning). With no options the resource is created or
+// GatedBy, OrphanWhen, Unowned, Auxiliary, BlockOnAbsence, IgnoreIfAbsent,
+// BlockOnForeignController, SuppressGraceInconsistencyWarning). With no options the resource is created or
 // updated and is required for the component to become Ready.
 //
 // A nil resource (a nil interface or a typed-nil pointer) is rejected with a
@@ -161,7 +161,10 @@ func (b *Builder) WithResource(resource Resource, opts ...ResourceOption) *Build
 	case options.Orphan:
 		b.component.orphanResources = append(b.component.orphanResources, resource)
 	case options.Delete:
-		b.component.deleteResources = append(b.component.deleteResources, resource)
+		b.component.deleteResources = append(b.component.deleteResources, reconcileEntry{
+			Resource: resource,
+			Options:  options,
+		})
 	default:
 		b.component.reconcileResources = append(b.component.reconcileResources, reconcileEntry{
 			Resource: resource,

pkg/component/builder_test.go

@@ -110,7 +110,7 @@ func TestBuilder_WithResource(t *testing.T) {
 	assert.True(t, comp.reconcileResources[1].Options.ReadOnly)
 
 	assert.Len(t, comp.deleteResources, 1)
-	assert.Equal(t, res3, comp.deleteResources[0])
+	assert.Equal(t, res3, comp.deleteResources[0].Resource)
 
 	assert.Len(t, comp.resourceLookup, 3)
 }

pkg/component/component.go

@@ -147,7 +147,7 @@ type Component struct {
 	// reconcileResources holds all non-delete resources in registration order.
 	// Each entry pairs the resource with its full options.
 	reconcileResources []reconcileEntry
-	deleteResources    []Resource
+	deleteResources    []reconcileEntry
 	orphanResources    []Resource
 	resourceLookup     map[string]Resource
 
@@ -494,18 +494,19 @@ func (c *Component) Reconcile(ctx context.Context, rec ReconcileContext) error {
 
 // allManagedResources returns every managed (non-read-only) resource known to
 // the component, combining non-read-only reconcile entries and delete entries
-// into a single slice. This is used when the feature gate is disabled and all
-// managed resources must be deleted. Read-only resources are excluded because
-// they are never created or modified by the component.
-func (c *Component) allManagedResources() []Resource {
-	resources := make([]Resource, 0, len(c.reconcileResources)+len(c.deleteResources))
+// into a single slice, each with its resolved options. This is used when the
+// feature gate is disabled and all managed resources must be deleted. Read-only
+// resources are excluded because they are never created or modified by the
+// component.
+func (c *Component) allManagedResources() []reconcileEntry {
+	entries := make([]reconcileEntry, 0, len(c.reconcileResources)+len(c.deleteResources))
 	for _, entry := range c.reconcileResources {
 		if !entry.Options.ReadOnly {
-			resources = append(resources, entry.Resource)
+			entries = append(entries, entry)
 		}
 	}
-	resources = append(resources, c.deleteResources...)
-	return resources
+	entries = append(entries, c.deleteResources...)
+	return entries
 }
 
 // prerequisiteBarrierActive reports whether the prerequisite initialization

pkg/component/component_test.go

@@ -582,7 +582,7 @@ var _ = Describe("Component Reconciler", func() {
 			res.On("Object").Return(cm, nil)
 			res.On("Identity").Return("ConfigMap/to-be-deleted")
 
-			comp.deleteResources = []Resource{res}
+			comp.deleteResources = []reconcileEntry{{Resource: res}}
 
 			// When
 			err := comp.Reconcile(ctx, recCtx)
@@ -648,7 +648,7 @@ var _ = Describe("Component Reconciler", func() {
 			res.On("Object").Return(nil, fmt.Errorf("delete object error"))
 			res.On("Identity").Return("failing-delete-resource")
 
-			comp.deleteResources = []Resource{res}
+			comp.deleteResources = []reconcileEntry{{Resource: res}}
 
 			// When
 			err := comp.Reconcile(ctx, recCtx)
@@ -709,7 +709,7 @@ var _ = Describe("Component Reconciler", func() {
 			delRes.On("Identity").Return("failing-suspend-delete-resource")
 
 			comp.reconcileResources = []reconcileEntry{{Resource: susRes}}
-			comp.deleteResources = []Resource{delRes}
+			comp.deleteResources = []reconcileEntry{{Resource: delRes}}
 
 			// When
 			err := comp.Reconcile(ctx, recCtx)

pkg/component/create.go

@@ -14,6 +14,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -316,6 +317,25 @@ func reconcileResources(
 			}
 		}
 
+		// A managed resource that another owner controls is blocked before the
+		// apply, so the forced apply never takes that owner's fields.
+		if entry.Options.BlockOnForeignController && !entry.Options.ReadOnly {
+			_, controller, err := observeController(ctx, rec, resource)
+			if err != nil {
+				return nil, err
+			}
+			if controller != nil {
+				results = append(results, reconcileResult{
+					Entry: entry,
+					Status: convergingStatusWithReason{
+						Status: convergingStatusGuardBlocked,
+						Reason: foreignControllerReason(controller),
+					},
+				})
+				return results, nil
+			}
+		}
+
 		// Process the resource based on its mode
 		var result *reconcileResult
 		var err error
@@ -360,6 +380,57 @@ func reconcileResources(
 	return results, nil
 }
 
+// observeController reads the live object of resource and returns it together
+// with its controller owner reference when that reference points at an owner
+// other than rec.Owner. The object is nil when it does not exist; the
+// controller is nil when the object has no controller reference or is
+// controlled by rec.Owner.
+//
+// The read goes through rec.APIReader when one is set and rec.Client otherwise.
+// The manager's Client serves reads from the informer cache, which can still
+// hold the object without the controller reference the API server already
+// carries; a forced apply decided on that stale read would take the other
+// owner's fields, which is what the option exists to stop.
+func observeController(
+	ctx context.Context, rec ReconcileContext, resource Resource,
+) (client.Object, *metav1.OwnerReference, error) {
+	obj, err := resource.Object()
+	if err != nil {
+		return nil, nil, fmt.Errorf(
+			"failed to retrieve object for resource %s: %w", resource.Identity(), err,
+		)
+	}
+	live, err := newEmptyObjectLike(obj)
+	if err != nil {
+		return nil, nil, fmt.Errorf(
+			"failed to prepare controller check for resource %s: %w", resource.Identity(), err,
+		)
+	}
+	var reader client.Reader = rec.Client
+	if rec.APIReader != nil {
+		reader = rec.APIReader
+	}
+	if err := reader.Get(ctx, client.ObjectKeyFromObject(obj), live); err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil, nil, nil
+		}
+		return nil, nil, fmt.Errorf(
+			"failed to read resource %s for controller check: %w", resource.Identity(), err,
+		)
+	}
+	controller := metav1.GetControllerOf(live)
+	if controller == nil || controller.UID == rec.Owner.GetUID() {
+		return live, nil, nil
+	}
+	return live, controller, nil
+}
+
+// foreignControllerReason is the blocked reason for an object controlled by
+// another owner, for example "controlled by DatabaseServer primary".
+func foreignControllerReason(controller *metav1.OwnerReference) string {
+	return fmt.Sprintf("controlled by %s %s", controller.Kind, controller.Name)
+}
+
 // mutateResource applies all desired-state mutations and sets the controller owner
 // reference. When skipOwnerRef is true the owner reference is intentionally omitted;
 // the resource is not garbage-collected when the owner CR is deleted.