🐛 fix: allow reconciliation of deadline-exceeded ClusterObjectSets

Previously, a watch predicate dropped update events for ClusterObjectSets
with ProgressDeadlineExceeded, preventing them from being archived or
cleaned up.

Remove the predicate and replace the inline deadline logic in Reconcile
with an enforceProgressDeadline() helper and a progressDeadline()
function that computes the absolute deadline from spec and metadata.

Add a deadline-aware rate limiter that caps exponential backoff at the
deadline and allows one immediate requeue when it passes, without
inspecting status conditions.

Add an e2e scenario that forces ProgressDeadlineExceeded, archives the
COS, and verifies resource cleanup.

Author: joelanford

internal/operator-controller/controllers/clusterobjectset_controller.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/utils/clock"
 	"pkg.package-operator.run/boxcutter"
 	"pkg.package-operator.run/boxcutter/machinery"
@@ -30,8 +31,8 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
-	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -80,30 +81,8 @@ func (c *ClusterObjectSetReconciler) Reconcile(ctx context.Context, req ctrl.Req
 
 	reconciledRev := existingRev.DeepCopy()
 	res, reconcileErr := c.reconcile(ctx, reconciledRev)
+	res = c.enforceProgressDeadline(reconciledRev, res, reconcileErr)
 
-	if pd := existingRev.Spec.ProgressDeadlineMinutes; pd > 0 {
-		cnd := meta.FindStatusCondition(reconciledRev.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing)
-		isStillProgressing := cnd != nil && cnd.Status == metav1.ConditionTrue && cnd.Reason != ocv1.ReasonSucceeded
-		succeeded := meta.IsStatusConditionTrue(reconciledRev.Status.Conditions, ocv1.ClusterObjectSetTypeSucceeded)
-		// check if we reached the progress deadline only if the revision is still progressing and has not succeeded yet
-		if isStillProgressing && !succeeded {
-			timeout := time.Duration(pd) * time.Minute
-			if c.Clock.Since(existingRev.CreationTimestamp.Time) > timeout {
-				// progress deadline reached, reset any errors and stop reconciling this revision
-				markAsNotProgressing(reconciledRev, ocv1.ReasonProgressDeadlineExceeded, fmt.Sprintf("Revision has not rolled out for %d minute(s).", pd))
-				reconcileErr = nil
-				res = ctrl.Result{}
-			} else if reconcileErr == nil {
-				// We want to requeue so far in the future that the next reconciliation
-				// can detect if the revision did not progress within the given timeout.
-				// Thus, we plan the next reconcile slightly after (+2secs) the timeout is passed.
-				drift := 2 * time.Second
-				requeueAfter := existingRev.CreationTimestamp.Time.Add(timeout).Add(drift).Sub(c.Clock.Now()).Round(time.Second)
-				l.Info(fmt.Sprintf("ProgressDeadline not exceeded, requeue after ~%v to check again.", requeueAfter))
-				res = ctrl.Result{RequeueAfter: requeueAfter}
-			}
-		}
-	}
 	// Do checks before any Update()s, as Update() may modify the resource structure!
 	updateStatus := !equality.Semantic.DeepEqual(existingRev.Status, reconciledRev.Status)
 
@@ -288,9 +267,7 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
 		} else {
 			markAsUnavailable(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
 		}
-		if meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing) == nil {
-			markAsProgressing(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
-		}
+		markAsProgressing(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
 	}
 
 	return ctrl.Result{}, nil
@@ -333,29 +310,19 @@ type Sourcoser interface {
 }
 
 func (c *ClusterObjectSetReconciler) SetupWithManager(mgr ctrl.Manager) error {
-	skipProgressDeadlineExceededPredicate := predicate.Funcs{
-		UpdateFunc: func(e event.UpdateEvent) bool {
-			rev, ok := e.ObjectNew.(*ocv1.ClusterObjectSet)
-			if !ok {
-				return true
-			}
-			// allow deletions to happen
-			if !rev.DeletionTimestamp.IsZero() {
-				return true
-			}
-			if cnd := meta.FindStatusCondition(rev.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing); cnd != nil && cnd.Status == metav1.ConditionFalse && cnd.Reason == ocv1.ReasonProgressDeadlineExceeded {
-				return false
-			}
-			return true
-		},
-	}
 	c.Clock = clock.RealClock{}
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(controller.Options{
+			RateLimiter: newDeadlineAwareRateLimiter(
+				workqueue.DefaultTypedControllerRateLimiter[ctrl.Request](),
+				mgr.GetClient(),
+				c.Clock,
+			),
+		}).
 		For(
 			&ocv1.ClusterObjectSet{},
 			builder.WithPredicates(
 				predicate.ResourceVersionChangedPredicate{},
-				skipProgressDeadlineExceededPredicate,
 			),
 		).
 		WatchesRawSource(
@@ -638,6 +605,60 @@ func setRetryingConditions(cos *ocv1.ClusterObjectSet, message string) {
 	}
 }
 
+// enforceProgressDeadline checks the progress deadline with a fresh clock
+// reading after the inner reconcile completes. If the deadline has passed
+// and probes have not all passed, it sets ProgressDeadlineExceeded. If the
+// deadline has not passed and there is no reconcile error, it caps RequeueAfter
+// to ensure a reconcile fires at the deadline.
+func (c *ClusterObjectSetReconciler) enforceProgressDeadline(cos *ocv1.ClusterObjectSet, res ctrl.Result, reconcileErr error) ctrl.Result {
+	deadline, ok := progressDeadline(cos)
+	if !ok {
+		return res
+	}
+	remaining := deadline.Sub(c.Clock.Now())
+	if remaining <= 0 {
+		markAsNotProgressing(cos, ocv1.ReasonProgressDeadlineExceeded,
+			fmt.Sprintf("Revision has not rolled out for %d minute(s).", cos.Spec.ProgressDeadlineMinutes))
+		return res
+	}
+	if remaining > 0 && reconcileErr == nil && (res.RequeueAfter == 0 || remaining < res.RequeueAfter) {
+		res.RequeueAfter = remaining
+	}
+	return res
+}
+
+// progressDeadline returns the absolute time at which the progress deadline
+// expires. It derives the deadline from spec and metadata only, with one
+// exception: it checks the Succeeded status condition so that a revision
+// recovering from drift is not penalised by the original deadline.
+//
+// Succeeded is a latch: there is no way to deduce from current cluster state
+// alone that a COS succeeded in the past. If Succeeded is removed or set to
+// False, this function will return a deadline and the reconciler will set
+// ProgressDeadlineExceeded even though the revision previously succeeded.
+//
+// Returns (zero, false) when there is no active deadline:
+//   - progressDeadlineMinutes is 0
+//   - the revision has already succeeded
+//   - the revision is archived (deadline is irrelevant)
+//   - the revision is being deleted
+func progressDeadline(cos *ocv1.ClusterObjectSet) (time.Time, bool) {
+	pd := cos.Spec.ProgressDeadlineMinutes
+	if pd <= 0 {
+		return time.Time{}, false
+	}
+	if meta.IsStatusConditionTrue(cos.Status.Conditions, ocv1.ClusterObjectSetTypeSucceeded) {
+		return time.Time{}, false
+	}
+	if cos.Spec.LifecycleState == ocv1.ClusterObjectSetLifecycleStateArchived {
+		return time.Time{}, false
+	}
+	if !cos.DeletionTimestamp.IsZero() {
+		return time.Time{}, false
+	}
+	return cos.CreationTimestamp.Time.Add(time.Duration(pd) * time.Minute), true
+}
+
 func markAsProgressing(cos *ocv1.ClusterObjectSet, reason, message string) {
 	meta.SetStatusCondition(&cos.Status.Conditions, metav1.Condition{
 		Type:               ocv1.ClusterObjectSetTypeProgressing,

internal/operator-controller/controllers/clusterobjectset_controller_test.go

@@ -988,7 +988,7 @@ func Test_ClusterObjectSetReconciler_Reconcile_ProgressDeadline(t *testing.T) {
 			revisionResult: &mockRevisionResult{
 				inTransition: true,
 			},
-			reconcileResult: ctrl.Result{RequeueAfter: 62 * time.Second},
+			reconcileResult: ctrl.Result{RequeueAfter: 60 * time.Second},
 			validate: func(t *testing.T, c client.Client) {
 				rev := &ocv1.ClusterObjectSet{}
 				err := c.Get(t.Context(), client.ObjectKey{

internal/operator-controller/controllers/deadline_ratelimiter.go

@@ -0,0 +1,76 @@
+//go:build !standard
+
+package controllers
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/clock"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	ocv1 "github.com/operator-framework/operator-controller/api/v1"
+)
+
+// deadlineAwareRateLimiter wraps a delegate rate limiter and caps the backoff
+// duration to the time remaining until the COS progress deadline, ensuring
+// that ProgressDeadlineExceeded is set promptly even during exponential backoff.
+//
+// After the deadline passes, it allows one immediate requeue (returning 0) so
+// the reconciler can set the ProgressDeadlineExceeded condition, then falls
+// back to the delegate's normal backoff. This avoids both tight-looping and
+// coupling to the COS's status conditions.
+type deadlineAwareRateLimiter struct {
+	delegate     workqueue.TypedRateLimiter[ctrl.Request]
+	client       client.Reader
+	clock        clock.Clock
+	pastDeadline sync.Map
+}
+
+func newDeadlineAwareRateLimiter(
+	delegate workqueue.TypedRateLimiter[ctrl.Request],
+	c client.Reader,
+	clk clock.Clock,
+) *deadlineAwareRateLimiter {
+	return &deadlineAwareRateLimiter{delegate: delegate, client: c, clock: clk}
+}
+
+func (r *deadlineAwareRateLimiter) When(item ctrl.Request) time.Duration {
+	backoff := r.delegate.When(item)
+
+	cos := &ocv1.ClusterObjectSet{}
+	if err := r.client.Get(context.Background(), item.NamespacedName, cos); err != nil {
+		return backoff
+	}
+
+	deadline, hasDeadline := progressDeadline(cos)
+	if !hasDeadline {
+		return backoff
+	}
+
+	remaining := deadline.Sub(r.clock.Now())
+	if remaining > 0 {
+		if remaining < backoff {
+			return remaining
+		}
+		return backoff
+	}
+
+	// Deadline has passed — allow one immediate requeue, then delegate.
+	if _, already := r.pastDeadline.LoadOrStore(item, struct{}{}); !already {
+		return 0
+	}
+	return backoff
+}
+
+func (r *deadlineAwareRateLimiter) Forget(item ctrl.Request) {
+	r.delegate.Forget(item)
+	r.pastDeadline.Delete(item)
+}
+
+func (r *deadlineAwareRateLimiter) NumRequeues(item ctrl.Request) int {
+	return r.delegate.NumRequeues(item)
+}