xds: suppress the duplicate errors at the client side

internal/xds/clients/xdsclient/authority.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -376,8 +377,12 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 					ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
 				})
 			}
+			isDuplicateErr := isDuplicateResourceErr(state.md.ErrState, uErr.Err)
 			state.md.ErrState = md.ErrState
 			state.md.Status = md.Status
+			if isDuplicateErr {
+				continue
+			}
 			for watcher := range state.watchers {
 				watcher := watcher
 				err := uErr.Err
@@ -987,3 +992,10 @@ func cacheState(r *resourceState) string {
 		return "requested"
 	}
 }
+
+func isDuplicateResourceErr(errState *xdsresource.UpdateErrorMetadata, resourceErr error) bool {
+	if errState == nil || errState.Err == nil || resourceErr == nil {
+		return false
+	}
+	return strings.Contains(errState.Err.Error(), resourceErr.Error())
+}

internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -506,3 +507,92 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+// TestADS_NACKError_DuplicateSuppression verifies that when a duplicate
+// invalid LDS resource is sent by the server, the client suppresses the
+// duplicate error notification to the watcher.
+func (s) TestADS_NACKError_DuplicateSuppression(t *testing.T) {
+	nackReceivedCh := make(chan struct{})
+	var nackCount atomic.Int32
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
+			if req.GetTypeUrl() != xdsresource.V3ListenerURL {
+				return nil
+			}
+			if req.GetErrorDetail() != nil {
+				if count := nackCount.Add(1); count == 3 {
+					close(nackReceivedCh)
+				}
+			}
+			return nil
+		},
+	})
+
+	const listenerName = "listener"
+	nodeID := uuid.New().String()
+
+	// Create an xDS client pointing to the above server.
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	client := createXDSClient(t, mgmtServer.Address, nodeID, grpctransport.NewBuilder(configs))
+
+	// Use a custom watcher that counts invocations.
+	errCh := testutils.NewChannelWithSize(1)
+	watcher := &countingListenerWatcher{errCh: errCh}
+	ldsCancel := client.WatchResource(xdsresource.V3ListenerURL, listenerName, watcher)
+	defer ldsCancel()
+
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{badListenerResource(t, listenerName)},
+		SkipValidation: true,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+
+	// Verify that the watcher receives the initial error.
+	v, err := errCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("timeout when waiting for an endpoints resource error from the management server: %v", err)
+	}
+	gotErr := v.(error)
+	wantNackErr := "no RouteSpecifier"
+	if !strings.Contains(gotErr.Error(), wantNackErr) {
+		t.Fatalf("update received with error: %v, want %q", gotErr, wantNackErr)
+	}
+
+	// Wait for the management server to receive at least 3 NACKs, indicating that
+	// the NACK resend loop has run multiple times.
+	select {
+	case <-nackReceivedCh:
+	case <-ctx.Done():
+		t.Fatalf("timeout waiting for 3 NACKs to be received by the server: %v", ctx.Err())
+	}
+
+	// Verify that the count of error callbacks is still exactly 1, proving that
+	// subsequent duplicate error updates were suppressed.
+	if count := watcher.errCount.Load(); count != 1 {
+		t.Fatalf("Error callback invoked %d times, want 1", count)
+	}
+}
+
+type countingListenerWatcher struct {
+	errCh    *testutils.Channel
+	errCount atomic.Int32
+}
+
+func (*countingListenerWatcher) ResourceChanged(_ xdsclient.ResourceData, done func()) {
+	done()
+}
+func (c *countingListenerWatcher) ResourceError(err error, done func()) {
+	c.errCount.Add(1)
+	c.errCh.Send(err)
+	done()
+}
+func (c *countingListenerWatcher) AmbientError(err error, done func()) {
+	c.errCount.Add(1)
+	c.errCh.Send(err)
+	done()
+}