Add AsyncOperationInProgress to ExternalObservation

When AsyncOperationInProgress is true, the managed reconciler sets
Synced=False with reason ReconcilePending instead of ReconcileSuccess.
This prevents a false Synced=True signal during long-running async
operations (e.g. MSK cluster instance type changes).

Fixes #941

Author: ajnye

apis/common/condition.go

@@ -63,6 +63,7 @@ const (
 	ReasonReconcileSuccess ConditionReason = "ReconcileSuccess"
 	ReasonReconcileError   ConditionReason = "ReconcileError"
 	ReasonReconcilePaused  ConditionReason = "ReconcilePaused"
+	ReasonReconcilePending ConditionReason = "ReconcilePending"
 )
 
 // See https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#typical-status-properties
@@ -312,3 +313,16 @@ func ReconcilePaused() Condition {
 		Reason:             ReasonReconcilePaused,
 	}
 }
+
+// ReconcilePending returns a condition indicating that reconciliation is
+// deferred pending an in-flight async operation. Unlike ReconcileError, this
+// does not trigger exponential backoff.
+func ReconcilePending(msg string) Condition {
+	return Condition{
+		Type:               TypeSynced,
+		Status:             corev1.ConditionFalse,
+		LastTransitionTime: metav1.Now(),
+		Reason:             ReasonReconcilePending,
+		Message:            msg,
+	}
+}

apis/common/v1/condition.go

@@ -60,6 +60,7 @@ const (
 	ReasonReconcileSuccess = common.ReasonReconcileSuccess
 	ReasonReconcileError   = common.ReasonReconcileError
 	ReasonReconcilePaused  = common.ReasonReconcilePaused
+	ReasonReconcilePending = common.ReasonReconcilePending
 )
 
 // See https://github.com/kubernetes/community/blob/master/contributors/devel/sig-architecture/api-conventions.md#typical-status-properties
@@ -134,3 +135,9 @@ func ReconcileError(err error) Condition {
 func ReconcilePaused() Condition {
 	return common.ReconcilePaused()
 }
+
+// ReconcilePending returns a condition indicating that reconciliation is
+// deferred pending an in-flight async operation.
+func ReconcilePending(msg string) Condition {
+	return common.ReconcilePending(msg)
+}

pkg/reconciler/managed/reconciler.go

@@ -526,6 +526,15 @@ type ExternalObservation struct {
 	// finding where the observed diverges from the desired state.
 	// The string should be a cmp.Diff that details the difference.
 	Diff string
+
+	// AsyncOperationInProgress indicates that an asynchronous operation
+	// (e.g. a long-running cloud API call) is currently in progress for
+	// this resource. When true, the managed reconciler will set
+	// Synced=False with reason ReconcilePending instead of
+	// ReconcileSuccess, and will not call Update(). This allows providers
+	// to suppress both Update() and the false ReconcileSuccess signal
+	// during async operations.
+	AsyncOperationInProgress bool
 }
 
 // An ExternalCreation is the result of the creation of an external resource.
@@ -1436,8 +1445,14 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu
 		// https://github.com/crossplane/crossplane/issues/289
 		reconcileAfter := r.pollIntervalHook(managed, r.pollInterval)
 		log.Debug("External resource is up to date", "requeue-after", time.Now().Add(reconcileAfter))
-		status.MarkConditions(xpv1.ReconcileSuccess())
-		r.metricRecorder.recordFirstTimeReady(managed)
+
+		if observation.AsyncOperationInProgress {
+			log.Debug("Async operation in progress, setting ReconcilePending")
+			status.MarkConditions(xpv1.ReconcilePending("Async operation in progress"))
+		} else {
+			status.MarkConditions(xpv1.ReconcileSuccess())
+			r.metricRecorder.recordFirstTimeReady(managed)
+		}
 
 		// record that we intentionally did not update the managed resource
 		// because no drift was detected. We call this so late in the reconcile

pkg/reconciler/managed/reconciler_legacy_test.go

@@ -1121,6 +1121,51 @@ func TestReconciler(t *testing.T) {
 			},
 			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
+		"ExternalResourceUpToDateAsyncOperationInProgress": {
+			reason: "When an async operation is in progress, Synced should be set to ReconcilePending instead of ReconcileSuccess.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: legacyManagedMockGetFn(nil, 42),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, obj client.Object, _ ...client.SubResourceUpdateOption) error {
+							want := newLegacyManaged(42)
+							want.SetConditions(xpv1.ReconcilePending("Async operation in progress").WithObservedGeneration(42))
+
+							if diff := cmp.Diff(want, obj, test.EquateConditions()); diff != "" {
+								reason := "An async-in-progress reconcile should set ReconcilePending, not ReconcileSuccess."
+								t.Errorf("\nReason: %s\n-want, +got:\n%s", reason, diff)
+							}
+
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.LegacyManaged{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.LegacyManaged{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnector(ExternalConnectorFn(func(_ context.Context, _ resource.Managed) (ExternalClient, error) {
+						c := &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{
+									ResourceExists:           true,
+									ResourceUpToDate:         true,
+									AsyncOperationInProgress: true,
+								}, nil
+							},
+							DisconnectFn: func(_ context.Context) error {
+								return nil
+							},
+						}
+
+						return c, nil
+					})),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+				},
+			},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
+		},
 		"ExternalResourceUpToDateWithJitter": {
 			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait with jitter added.",
 			args: args{

pkg/reconciler/managed/reconciler_modern_test.go

@@ -1127,6 +1127,51 @@ func TestModernReconciler(t *testing.T) {
 			},
 			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
+		"ExternalResourceUpToDateAsyncOperationInProgress": {
+			reason: "When an async operation is in progress, Synced should be set to ReconcilePending instead of ReconcileSuccess.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: modernManagedMockGetFn(nil, 42),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, obj client.Object, _ ...client.SubResourceUpdateOption) error {
+							want := newModernManaged(42)
+							want.SetConditions(xpv1.ReconcilePending("Async operation in progress").WithObservedGeneration(42))
+
+							if diff := cmp.Diff(want, obj, test.EquateConditions()); diff != "" {
+								reason := "An async-in-progress reconcile should set ReconcilePending, not ReconcileSuccess."
+								t.Errorf("\nReason: %s\n-want, +got:\n%s", reason, diff)
+							}
+
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.ModernManaged{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.ModernManaged{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnector(ExternalConnectorFn(func(_ context.Context, _ resource.Managed) (ExternalClient, error) {
+						c := &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{
+									ResourceExists:           true,
+									ResourceUpToDate:         true,
+									AsyncOperationInProgress: true,
+								}, nil
+							},
+							DisconnectFn: func(_ context.Context) error {
+								return nil
+							},
+						}
+
+						return c, nil
+					})),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+				},
+			},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
+		},
 		"ExternalResourceUpToDateWithJitter": {
 			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait with jitter added.",
 			args: args{