xds: implement server feature fail_on_data_errors (#12544)

Implements server feature `fail_on_data_errors`, per gRFC A88:
https://github.com/grpc/proposal/blob/master/A88-xds-data-error-handling.md

Author: shivaspeaks

xds/src/main/java/io/grpc/xds/client/Bootstrapper.java

@@ -65,18 +65,22 @@ public abstract static class ServerInfo {
 
     public abstract boolean resourceTimerIsTransientError();
 
+    public abstract boolean failOnDataErrors();
+
     @VisibleForTesting
     public static ServerInfo create(String target, @Nullable Object implSpecificConfig) {
       return new AutoValue_Bootstrapper_ServerInfo(target, implSpecificConfig,
-          false, false, false);
+          false, false, false, false);
     }
 
     @VisibleForTesting
     public static ServerInfo create(
-        String target, Object implSpecificConfig, boolean ignoreResourceDeletion,
-        boolean isTrustedXdsServer, boolean resourceTimerIsTransientError) {
+        String target, Object implSpecificConfig,
+        boolean ignoreResourceDeletion, boolean isTrustedXdsServer,
+        boolean resourceTimerIsTransientError, boolean failOnDataErrors) {
       return new AutoValue_Bootstrapper_ServerInfo(target, implSpecificConfig,
-          ignoreResourceDeletion, isTrustedXdsServer, resourceTimerIsTransientError);
+          ignoreResourceDeletion, isTrustedXdsServer,
+          resourceTimerIsTransientError, failOnDataErrors);
     }
   }
 

xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java

@@ -58,6 +58,7 @@ public abstract class BootstrapperImpl extends Bootstrapper {
   private static final String SERVER_FEATURE_TRUSTED_XDS_SERVER = "trusted_xds_server";
   private static final String
       SERVER_FEATURE_RESOURCE_TIMER_IS_TRANSIENT_ERROR = "resource_timer_is_transient_error";
+  private static final String SERVER_FEATURE_FAIL_ON_DATA_ERRORS = "fail_on_data_errors";
 
   @VisibleForTesting
   static boolean enableXdsFallback = GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_FALLBACK, true);
@@ -257,6 +258,7 @@ private List<ServerInfo> parseServerInfos(List<?> rawServerConfigs, XdsLogger lo
 
       boolean resourceTimerIsTransientError = false;
       boolean ignoreResourceDeletion = false;
+      boolean failOnDataErrors = false;
       // "For forward compatibility reasons, the client will ignore any entry in the list that it
       // does not understand, regardless of type."
       List<?> serverFeatures = JsonUtil.getList(serverConfig, "server_features");
@@ -267,12 +269,14 @@ private List<ServerInfo> parseServerInfos(List<?> rawServerConfigs, XdsLogger lo
         }
         resourceTimerIsTransientError = xdsDataErrorHandlingEnabled
             && serverFeatures.contains(SERVER_FEATURE_RESOURCE_TIMER_IS_TRANSIENT_ERROR);
+        failOnDataErrors = xdsDataErrorHandlingEnabled
+            && serverFeatures.contains(SERVER_FEATURE_FAIL_ON_DATA_ERRORS);
       }
       servers.add(
           ServerInfo.create(serverUri, implSpecificConfig, ignoreResourceDeletion,
               serverFeatures != null
                   && serverFeatures.contains(SERVER_FEATURE_TRUSTED_XDS_SERVER),
-              resourceTimerIsTransientError));
+              resourceTimerIsTransientError, failOnDataErrors));
     }
     return servers.build();
   }

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -603,16 +603,20 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
       }
 
       if (invalidResources.contains(resourceName)) {
-        // The resource update is invalid. Capture the error without notifying the watchers.
+        // The resource update is invalid (NACK). Handle as a data error.
         subscriber.onRejected(args.versionInfo, updateTime, errorDetail);
-      }
-
-      if (invalidResources.contains(resourceName)) {
-        // The resource is missing. Reuse the cached resource if possible.
-        if (subscriber.data == null) {
-          // No cached data. Notify the watchers of an invalid update.
-          subscriber.onError(Status.UNAVAILABLE.withDescription(errorDetail), processingTracker);
+        
+        // Handle data errors (NACKs) based on fail_on_data_errors server feature.
+        // When xdsDataErrorHandlingEnabled is true and fail_on_data_errors is present,
+        // delete cached data so onError will call onResourceChanged instead of onAmbientError.
+        // When xdsDataErrorHandlingEnabled is false, use old behavior (always keep cached data).
+        if (BootstrapperImpl.xdsDataErrorHandlingEnabled && subscriber.data != null
+            && args.serverInfo.failOnDataErrors()) {
+          subscriber.data = null;
         }
+        // Call onError, which will decide whether to call onResourceChanged or onAmbientError
+        // based on whether data exists after the above deletion.
+        subscriber.onError(Status.UNAVAILABLE.withDescription(errorDetail), processingTracker);
         continue;
       }
 
@@ -866,20 +870,42 @@ void onAbsent(@Nullable ProcessingTracker processingTracker, ServerInfo serverIn
         return;
       }
 
-      // Ignore deletion of State of the World resources when this feature is on,
-      // and the resource is reusable.
+      // Handle data errors (resource deletions) based on fail_on_data_errors server feature.
+      // When xdsDataErrorHandlingEnabled is true and fail_on_data_errors is not present,
+      // we treat deletions as ambient errors and keep using the cached resource.
+      // When fail_on_data_errors is present, we delete the cached resource and fail.
+      // When xdsDataErrorHandlingEnabled is false, use the old behavior (ignore_resource_deletion).
       boolean ignoreResourceDeletionEnabled = serverInfo.ignoreResourceDeletion();
-      if (ignoreResourceDeletionEnabled && type.isFullStateOfTheWorld() && data != null) {
-        if (!resourceDeletionIgnored) {
-          logger.log(XdsLogLevel.FORCE_WARNING,
-              "xds server {0}: ignoring deletion for resource type {1} name {2}}",
-              serverInfo.target(), type, resource);
-          resourceDeletionIgnored = true;
+      boolean failOnDataErrors = serverInfo.failOnDataErrors();
+      boolean xdsDataErrorHandlingEnabled = BootstrapperImpl.xdsDataErrorHandlingEnabled;
+
+      if (type.isFullStateOfTheWorld() && data != null) {
+        // New behavior (per gRFC A88): Default is to treat deletions as ambient errors
+        if (xdsDataErrorHandlingEnabled && !failOnDataErrors) {
+          if (!resourceDeletionIgnored) {
+            logger.log(XdsLogLevel.FORCE_WARNING,
+                "xds server {0}: ignoring deletion for resource type {1} name {2}}",
+                serverInfo.target(), type, resource);
+            resourceDeletionIgnored = true;
+          }
+          Status deletionStatus = Status.NOT_FOUND.withDescription(
+              "Resource " + resource + " deleted from server");
+          onAmbientError(deletionStatus, processingTracker);
+          return;
+        }
+        // Old behavior: Use ignore_resource_deletion server feature
+        if (!xdsDataErrorHandlingEnabled && ignoreResourceDeletionEnabled) {
+          if (!resourceDeletionIgnored) {
+            logger.log(XdsLogLevel.FORCE_WARNING,
+                "xds server {0}: ignoring deletion for resource type {1} name {2}}",
+                serverInfo.target(), type, resource);
+            resourceDeletionIgnored = true;
+          }
+          Status deletionStatus = Status.NOT_FOUND.withDescription(
+              "Resource " + resource + " deleted from server");
+          onAmbientError(deletionStatus, processingTracker);
+          return;
         }
-        Status deletionStatus = Status.NOT_FOUND.withDescription(
-            "Resource " + resource + " deleted from server");
-        onAmbientError(deletionStatus, processingTracker);
-        return;
       }
 
       logger.log(XdsLogLevel.INFO, "Conclude {0} resource {1} not exist", type, resource);

xds/src/test/java/io/grpc/xds/GrpcBootstrapperImplTest.java

@@ -723,6 +723,52 @@ public void serverFeatures_ignoresUnknownValues() throws XdsInitializationExcept
     assertThat(serverInfo.isTrustedXdsServer()).isTrue();
   }
 
+  @Test
+  public void serverFeature_failOnDataErrors() throws XdsInitializationException {
+    BootstrapperImpl.xdsDataErrorHandlingEnabled = true;
+    String rawData = "{\n"
+        + "  \"xds_servers\": [\n"
+        + "    {\n"
+        + "      \"server_uri\": \"" + SERVER_URI + "\",\n"
+        + "      \"channel_creds\": [\n"
+        + "        {\"type\": \"insecure\"}\n"
+        + "      ],\n"
+        + "      \"server_features\": [\"fail_on_data_errors\"]\n"
+        + "    }\n"
+        + "  ]\n"
+        + "}";
+
+    bootstrapper.setFileReader(createFileReader(BOOTSTRAP_FILE_PATH, rawData));
+    BootstrapInfo info = bootstrapper.bootstrap();
+    ServerInfo serverInfo = Iterables.getOnlyElement(info.servers());
+    assertThat(serverInfo.target()).isEqualTo(SERVER_URI);
+    assertThat(serverInfo.implSpecificConfig()).isInstanceOf(InsecureChannelCredentials.class);
+    assertThat(serverInfo.failOnDataErrors()).isTrue();
+    BootstrapperImpl.xdsDataErrorHandlingEnabled = false;
+  }
+
+  @Test
+  public void serverFeature_failOnDataErrors_requiresEnvVar() throws XdsInitializationException {
+    BootstrapperImpl.xdsDataErrorHandlingEnabled = false;
+    String rawData = "{\n"
+        + "  \"xds_servers\": [\n"
+        + "    {\n"
+        + "      \"server_uri\": \"" + SERVER_URI + "\",\n"
+        + "      \"channel_creds\": [\n"
+        + "        {\"type\": \"insecure\"}\n"
+        + "      ],\n"
+        + "      \"server_features\": [\"fail_on_data_errors\"]\n"
+        + "    }\n"
+        + "  ]\n"
+        + "}";
+
+    bootstrapper.setFileReader(createFileReader(BOOTSTRAP_FILE_PATH, rawData));
+    BootstrapInfo info = bootstrapper.bootstrap();
+    ServerInfo serverInfo = Iterables.getOnlyElement(info.servers());
+    // Should be false when env var is not enabled
+    assertThat(serverInfo.failOnDataErrors()).isFalse();
+  }
+
   @Test
   public void notFound() {
     bootstrapper.bootstrapPathFromEnvVar = null;

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java

@@ -3614,7 +3614,7 @@ private static Filter buildHttpConnectionManagerFilter(HttpFilter... httpFilters
 
   private XdsResourceType.Args getXdsResourceTypeArgs(boolean isTrustedServer) {
     return new XdsResourceType.Args(
-        ServerInfo.create("http://td", "", false, isTrustedServer, false), "1.0", null, null, null, null
+        ServerInfo.create("http://td", "", false, isTrustedServer, false, false), "1.0", null, null, null, null
     );
   }
 }