ScheduleRunOnce if not already scheduled

controller/controller.go

@@ -242,7 +242,11 @@ func filterARecords(endpoints []*endpoint.Endpoint) []string {
 func (c *Controller) ScheduleRunOnce(now time.Time) {
 	c.nextRunAtMux.Lock()
 	defer c.nextRunAtMux.Unlock()
-	c.nextRunAt = now.Add(c.MinEventSyncInterval)
+	// Shedule only if a reconciliation is not already planned
+	// to happen in the following c.MinEventSyncInterval
+	if !c.nextRunAt.Before(now.Add(c.MinEventSyncInterval)) {
+		c.nextRunAt = now.Add(c.MinEventSyncInterval)
+	}
 }
 
 func (c *Controller) ShouldRunOnce(now time.Time) bool {

controller/controller_test.go

@@ -19,12 +19,13 @@ package controller
 import (
 	"context"
 	"errors"
-	"github.com/prometheus/client_golang/prometheus"
 	"math"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/internal/testutils"
 	"sigs.k8s.io/external-dns/pkg/apis/externaldns"
@@ -236,6 +237,17 @@ func TestShouldRunOnce(t *testing.T) {
 
 	// But not two times
 	assert.False(t, ctrl.ShouldRunOnce(now))
+
+	// Multiple ingresses or services changes, closer than MinInterval from each other
+	firstChangeTime := now
+	secondChangeTime := firstChangeTime.Add(time.Second)
+	// First change
+	ctrl.ScheduleRunOnce(firstChangeTime)
+	// Second change
+	ctrl.ScheduleRunOnce(secondChangeTime)
+	// Should not postpone the reconciliation further than firstChangeTime + MinInterval
+	now = now.Add(ctrl.MinEventSyncInterval)
+	assert.True(t, ctrl.ShouldRunOnce(now))
 }
 
 func testControllerFiltersDomains(t *testing.T, configuredEndpoints []*endpoint.Endpoint, domainFilter endpoint.DomainFilterInterface, providerEndpoints []*endpoint.Endpoint, expectedChanges []*plan.Changes) {