fix(broker): reconcile ingress contracts idempotently (#797)

Signed-off-by: kahirokunn <okinakahiro@gmail.com>

Author: kahirokunn

pkg/broker/contract/manager.go

@@ -61,6 +61,13 @@ func (m *Manager) GetContract(ctx context.Context) (*Contract, error) {
 
 // UpdateBroker adds or updates a broker in the contract ConfigMap
 func (m *Manager) UpdateBroker(ctx context.Context, broker BrokerContract) error {
+	_, err := m.UpdateBrokerIfChanged(ctx, broker)
+	return err
+}
+
+// UpdateBrokerIfChanged adds or updates a broker and reports whether it wrote
+// the contract ConfigMap.
+func (m *Manager) UpdateBrokerIfChanged(ctx context.Context, broker BrokerContract) (bool, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -74,7 +81,7 @@ func (m *Manager) UpdateBroker(ctx context.Context, broker BrokerContract) error
 
 			data, err := SerializeContract(contract)
 			if err != nil {
-				return err
+				return false, err
 			}
 
 			cm = &corev1.ConfigMap{
@@ -94,22 +101,24 @@ func (m *Manager) UpdateBroker(ctx context.Context, broker BrokerContract) error
 			}
 
 			_, err = m.client.CoreV1().ConfigMaps(m.namespace).Create(ctx, cm, metav1.CreateOptions{})
-			return err
+			return err == nil, err
 		}
-		return err
+		return false, err
 	}
 
 	// Update existing ConfigMap
 	contract, err := ParseContract(cm)
 	if err != nil {
-		return err
+		return false, err
 	}
 
-	contract.SetBroker(broker)
+	if !contract.SetBrokerIfChanged(broker) {
+		return false, nil
+	}
 
 	data, err := SerializeContract(contract)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	if cm.Data == nil {
@@ -123,7 +132,7 @@ func (m *Manager) UpdateBroker(ctx context.Context, broker BrokerContract) error
 	cm.Annotations[ContractGenerationAnnotation] = fmt.Sprintf("%d", contract.Generation)
 
 	_, err = m.client.CoreV1().ConfigMaps(m.namespace).Update(ctx, cm, metav1.UpdateOptions{})
-	return err
+	return err == nil, err
 }
 
 // DeleteBroker removes a broker from the contract ConfigMap

pkg/broker/contract/manager_test.go

@@ -274,6 +274,88 @@ func TestManager_UpdateBroker_AnnotationUpdated(t *testing.T) {
 	}
 }
 
+func TestManager_UpdateBroker_SteadyStateDoesNotWrite(t *testing.T) {
+	existingCM := contractConfigMap(&Contract{Brokers: make(map[string]BrokerContract)})
+	client := fake.NewClientset(existingCM)
+	manager := &Manager{
+		client:    client,
+		lister:    newFakeConfigMapLister(),
+		namespace: testSystemNamespace,
+	}
+	broker := BrokerContract{
+		UID: "uid-1", Namespace: "ns", Name: "br", StreamName: "stream-1",
+		PublishSubject: "broker.ns.br", Path: "/ns/br", Generation: 4,
+	}
+	countWrites := func() int {
+		writes := 0
+		for _, action := range client.Actions() {
+			if action.GetResource().Resource == "configmaps" &&
+				(action.GetVerb() == "create" || action.GetVerb() == "update") {
+				writes++
+			}
+		}
+		return writes
+	}
+	getContract := func() *Contract {
+		t.Helper()
+		cm, err := client.CoreV1().ConfigMaps(testSystemNamespace).Get(
+			context.Background(), ConfigMapName, metav1.GetOptions{},
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+		contract, err := ParseContract(cm)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return contract
+	}
+
+	wrote, err := manager.UpdateBrokerIfChanged(context.Background(), broker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !wrote {
+		t.Fatal("initial UpdateBrokerIfChanged() changed = false, want true")
+	}
+	if got := getContract().Generation; got != 1 {
+		t.Fatalf("Generation after initial UpdateBroker() = %d, want 1", got)
+	}
+	if got := countWrites(); got != 1 {
+		t.Fatalf("ConfigMap writes after initial UpdateBroker() = %d, want 1", got)
+	}
+
+	wrote, err = manager.UpdateBrokerIfChanged(context.Background(), broker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if wrote {
+		t.Fatal("identical UpdateBrokerIfChanged() changed = true, want false")
+	}
+	if got := getContract().Generation; got != 1 {
+		t.Fatalf("Generation after identical UpdateBroker() = %d, want unchanged 1", got)
+	}
+	if got := countWrites(); got != 1 {
+		t.Fatalf("ConfigMap writes after identical UpdateBroker() = %d, want unchanged 1", got)
+	}
+
+	changedContract := broker
+	changedContract.PublishSubject = "broker.ns.br.changed"
+	wrote, err = manager.UpdateBrokerIfChanged(context.Background(), changedContract)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !wrote {
+		t.Fatal("changed UpdateBrokerIfChanged() changed = false, want true")
+	}
+	if got := getContract().Generation; got != 2 {
+		t.Fatalf("Generation after changed UpdateBroker() = %d, want 2", got)
+	}
+	if got := countWrites(); got != 2 {
+		t.Fatalf("ConfigMap writes after changed UpdateBroker() = %d, want 2", got)
+	}
+}
+
 // --- DeleteBroker tests ---
 
 func TestManager_DeleteBroker_ConfigMapNotFound(t *testing.T) {

pkg/broker/contract/types.go

@@ -88,13 +88,24 @@ func (c *Contract) GetBrokerByPath(path string) (BrokerContract, bool) {
 	return BrokerContract{}, false
 }
 
-// SetBroker adds or updates a broker in the contract
+// SetBroker adds or updates a broker in the contract.
 func (c *Contract) SetBroker(broker BrokerContract) {
+	c.SetBrokerIfChanged(broker)
+}
+
+// SetBrokerIfChanged adds or updates a broker and reports whether the
+// serialized contract changed.
+func (c *Contract) SetBrokerIfChanged(broker BrokerContract) bool {
 	if c.Brokers == nil {
 		c.Brokers = make(map[string]BrokerContract)
 	}
-	c.Brokers[BrokerKey(broker.Namespace, broker.Name)] = broker
+	key := BrokerKey(broker.Namespace, broker.Name)
+	if existing, ok := c.Brokers[key]; ok && existing == broker {
+		return false
+	}
+	c.Brokers[key] = broker
 	c.Generation++
+	return true
 }
 
 // DeleteBroker removes a broker from the contract

pkg/broker/contract/types_test.go

@@ -129,6 +129,39 @@ func TestContract_SetBroker(t *testing.T) {
 			t.Errorf("StreamName = %q, want %q", got, "new")
 		}
 	})
+
+	t.Run("identical broker is a steady-state no-op", func(t *testing.T) {
+		broker := BrokerContract{
+			UID: "uid-1", Namespace: "ns", Name: "br", StreamName: "stream-1",
+			PublishSubject: "broker.ns.br", Path: "/ns/br", Generation: 4,
+		}
+		contract := &Contract{Brokers: make(map[string]BrokerContract)}
+
+		if changed := contract.SetBrokerIfChanged(broker); !changed {
+			t.Fatal("initial SetBrokerIfChanged() changed = false, want true")
+		}
+		if contract.Generation != 1 {
+			t.Fatalf("Generation after initial SetBroker() = %d, want 1", contract.Generation)
+		}
+		if changed := contract.SetBrokerIfChanged(broker); changed {
+			t.Fatal("identical SetBrokerIfChanged() changed = true, want false")
+		}
+		if contract.Generation != 1 {
+			t.Fatalf("Generation after identical SetBroker() = %d, want unchanged 1", contract.Generation)
+		}
+
+		changed := broker
+		changed.StreamName = "stream-2"
+		if changedContract := contract.SetBrokerIfChanged(changed); !changedContract {
+			t.Fatal("changed SetBrokerIfChanged() changed = false, want true")
+		}
+		if contract.Generation != 2 {
+			t.Fatalf("Generation after changed SetBroker() = %d, want 2", contract.Generation)
+		}
+		if got := contract.Brokers["ns/br"].StreamName; got != "stream-2" {
+			t.Fatalf("stored StreamName = %q, want %q", got, "stream-2")
+		}
+	})
 }
 
 func TestContract_DeleteBroker(t *testing.T) {

pkg/broker/controller/reconciler.go

@@ -139,14 +139,17 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, b *eventingv1.Broker) pk
 	}
 
 	// Step 4: Reconcile ingress service
-	if err := r.contractManager.UpdateBroker(ctx, brokerContract); err != nil {
+	contractUpdated, err := r.contractManager.UpdateBrokerIfChanged(ctx, brokerContract)
+	if err != nil {
 		logger.Errorw("Failed to update contract", zap.Error(err))
 		controller.GetEventRecorder(ctx).Event(b, corev1.EventTypeWarning, ReasonContractFailed, err.Error())
 		b.Status.MarkIngressFailed("ContractUpdateFailed", "Failed to update contract ConfigMap: %v", err)
 		return fmt.Errorf("failed to update contract: %w", err)
 	}
 
-	controller.GetEventRecorder(ctx).Event(b, corev1.EventTypeNormal, ReasonContractUpdated, "Contract updated")
+	if contractUpdated {
+		controller.GetEventRecorder(ctx).Event(b, corev1.EventTypeNormal, ReasonContractUpdated, "Contract updated")
+	}
 
 	// Step 4: Check shared ingress deployment readiness
 	if err := r.propagateIngressAvailability(ctx, b); err != nil {