xds: Add fallback_on_reachability_only authority knob (A95)

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

@@ -128,11 +128,28 @@ public abstract static class AuthorityInfo {
      */
     public abstract ImmutableList<ServerInfo> xdsServers();
 
+    /**
+     * Whether to fallback to the next xDS server based solely on the reachability of the primary
+     * server, as described in gRFC A95.
+     *
+     * <p>If {@code false} (the default), fallback happens only if the primary server is unreachable
+     * <i>and</i> there are uncached resources, as described in gRFC A71. If {@code true}, fallback
+     * happens whenever the primary server is unreachable, even if all resources are cached.
+     */
+    public abstract boolean fallbackOnReachabilityOnly();
+
     public static AuthorityInfo create(
         String clientListenerResourceNameTemplate, List<ServerInfo> xdsServers) {
+      return create(clientListenerResourceNameTemplate, xdsServers, false);
+    }
+
+    public static AuthorityInfo create(
+        String clientListenerResourceNameTemplate, List<ServerInfo> xdsServers,
+        boolean fallbackOnReachabilityOnly) {
       checkArgument(!xdsServers.isEmpty(), "xdsServers must not be empty");
       return new AutoValue_Bootstrapper_AuthorityInfo(
-          clientListenerResourceNameTemplate, ImmutableList.copyOf(xdsServers));
+          clientListenerResourceNameTemplate, ImmutableList.copyOf(xdsServers),
+          fallbackOnReachabilityOnly);
     }
   }
 

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

@@ -47,6 +47,8 @@ public abstract class BootstrapperImpl extends Bootstrapper {
       "GRPC_EXPERIMENTAL_XDS_FALLBACK";
   public static final String GRPC_EXPERIMENTAL_XDS_DATA_ERROR_HANDLING =
       "GRPC_EXPERIMENTAL_XDS_DATA_ERROR_HANDLING";
+  public static final String GRPC_EXPERIMENTAL_XDS_ENDPOINT_FALLBACK =
+      "GRPC_EXPERIMENTAL_XDS_ENDPOINT_FALLBACK";
 
   // Client features.
   @VisibleForTesting
@@ -65,6 +67,10 @@ public abstract class BootstrapperImpl extends Bootstrapper {
   @VisibleForTesting
   static boolean enableXdsFallback = GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_FALLBACK, true);
 
+  @VisibleForTesting
+  public static boolean enableEndpointFallback =
+      GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_ENDPOINT_FALLBACK, false);
+
   @VisibleForTesting
   public static boolean xdsDataErrorHandlingEnabled
       = GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_DATA_ERROR_HANDLING, false);
@@ -235,8 +241,13 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
           }
           authorityServers = parseServerInfos(rawAuthorityServers, logger);
         }
+        // gRFC A95: fall back based solely on primary server reachability.
+        boolean fallbackOnReachabilityOnly = enableEndpointFallback
+            && Boolean.TRUE.equals(
+                JsonUtil.getBoolean(rawAuthority, "fallback_on_reachability_only"));
         authorityInfoMapBuilder.put(
-            authorityName, AuthorityInfo.create(clientListnerTemplate, authorityServers));
+            authorityName, AuthorityInfo.create(
+                clientListnerTemplate, authorityServers, fallbackOnReachabilityOnly));
       }
       builder.authorities(authorityInfoMapBuilder.buildOrThrow());
     }

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

@@ -556,6 +556,19 @@ private ImmutableList<ServerInfo> getServerInfos(String authority) {
     }
   }
 
+  /**
+   * Whether the given authority is configured to fall back based solely on the reachability of the
+   * primary server, as described in gRFC A95. The knob is only defined for named authorities, so
+   * resources served by the top-level server list always use the gRFC A71 behavior.
+   */
+  private boolean fallbackOnReachabilityOnly(@Nullable String authority) {
+    if (authority == null) {
+      return false;
+    }
+    AuthorityInfo authorityInfo = bootstrapInfo.authorities().get(authority);
+    return authorityInfo != null && authorityInfo.fallbackOnReachabilityOnly();
+  }
+
   @SuppressWarnings("unchecked")
   private <T extends ResourceUpdate> void handleResourceUpdate(
       XdsResourceType.Args args, List<Any> resources, XdsResourceType<T> xdsResourceType,
@@ -1057,8 +1070,10 @@ public void handleStreamClosed(Status status, boolean shouldTryFallback) {
           if (!authoritiesForClosedCpc.contains(subscriber.authority)) {
             continue;
           }
-          // If subscriber already has data, this is an ambient error.
-          if (subscriber.hasResult()) {
+          // If subscriber already has data, this is an ambient error. But if the authority is
+          // configured to fall back based solely on reachability (gRFC A95), still attempt
+          // fallback below even though the resource is cached.
+          if (subscriber.hasResult() && !fallbackOnReachabilityOnly(subscriber.authority)) {
             subscriber.onError(status, null);
             continue;
           }

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

@@ -22,6 +22,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verifyNoInteractions;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
 import io.grpc.InsecureChannelCredentials;
@@ -62,11 +63,14 @@ public class GrpcBootstrapperImplTest {
   private String originalBootstrapConfigFromEnvVar;
   private String originalBootstrapConfigFromSysProp;
   private boolean originalExperimentalXdsFallbackFlag;
+  private boolean originalEndpointFallbackFlag;
 
   @Before
   public void setUp() {
     saveEnvironment();
     originalExperimentalXdsFallbackFlag = CommonBootstrapperTestUtils.setEnableXdsFallback(true);
+    originalEndpointFallbackFlag =
+        CommonBootstrapperTestUtils.setEnableEndpointFallback(false);
     bootstrapper.bootstrapPathFromEnvVar = BOOTSTRAP_FILE_PATH;
   }
 
@@ -84,6 +88,7 @@ public void restoreEnvironment() {
     bootstrapper.bootstrapConfigFromEnvVar = originalBootstrapConfigFromEnvVar;
     bootstrapper.bootstrapConfigFromSysProp = originalBootstrapConfigFromSysProp;
     CommonBootstrapperTestUtils.setEnableXdsFallback(originalExperimentalXdsFallbackFlag);
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(originalEndpointFallbackFlag);
   }
 
   @Test
@@ -1007,6 +1012,59 @@ public void parseAuthorities() throws Exception {
         .isEqualTo("xdstp://a.com/envoy.config.listener.v3.Listener/%s");
     assertThat(authorityInfo.xdsServers()).hasSize(1);
     assertThat(authorityInfo.xdsServers().get(0).target()).isEqualTo("td2.googleapis.com:443");
+    // gRFC A95: defaults to false when not specified.
+    assertThat(authorityInfo.fallbackOnReachabilityOnly()).isFalse();
+  }
+
+  @Test
+  public void parseAuthorities_fallbackOnReachabilityOnly() throws Exception {
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(true);
+    bootstrapper.setFileReader(
+        createFileReader(BOOTSTRAP_FILE_PATH, buildAuthorityBootstrap(true)));
+    BootstrapInfo info = bootstrapper.bootstrap();
+    assertThat(info.authorities().get("a.com").fallbackOnReachabilityOnly()).isTrue();
+
+    bootstrapper.setFileReader(
+        createFileReader(BOOTSTRAP_FILE_PATH, buildAuthorityBootstrap(false)));
+    info = bootstrapper.bootstrap();
+    assertThat(info.authorities().get("a.com").fallbackOnReachabilityOnly()).isFalse();
+  }
+
+  @Test
+  public void parseAuthorities_fallbackOnReachabilityOnly_ignoredWhenEnvVarDisabled()
+      throws Exception {
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(false);
+    bootstrapper.setFileReader(
+        createFileReader(BOOTSTRAP_FILE_PATH, buildAuthorityBootstrap(true)));
+    BootstrapInfo info = bootstrapper.bootstrap();
+    assertThat(info.authorities().get("a.com").fallbackOnReachabilityOnly()).isFalse();
+  }
+
+  @Test
+  public void authorityInfo_defaultsFallbackOnReachabilityOnlyToFalse() {
+    AuthorityInfo authorityInfo = AuthorityInfo.create(
+        "xdstp://a.com/envoy.config.listener.v3.Listener/%s",
+        ImmutableList.of(ServerInfo.create(SERVER_URI, InsecureChannelCredentials.create())));
+    assertThat(authorityInfo.fallbackOnReachabilityOnly()).isFalse();
+  }
+
+  private static String buildAuthorityBootstrap(boolean fallbackOnReachabilityOnly) {
+    return "{\n"
+        + "  \"authorities\": {\n"
+        + "    \"a.com\": {\n"
+        + "      \"client_listener_resource_name_template\": \"xdstp://a.com/v1.Listener/id-%s\",\n"
+        + "      \"fallback_on_reachability_only\": " + fallbackOnReachabilityOnly + "\n"
+        + "    }\n"
+        + "  },\n"
+        + "  \"xds_servers\": [\n"
+        + "    {\n"
+        + "      \"server_uri\": \"" + SERVER_URI + "\",\n"
+        + "      \"channel_creds\": [\n"
+        + "        {\"type\": \"insecure\"}\n"
+        + "      ]\n"
+        + "    }\n"
+        + "  ]\n"
+        + "}";
   }
 
   @Test

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

@@ -18,6 +18,7 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
+import static io.grpc.xds.XdsTestControlPlaneService.ADS_TYPE_URL_LDS;
 import static org.mockito.AdditionalAnswers.delegatesTo;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
@@ -86,6 +87,9 @@ public class XdsClientFallbackTest {
   private static final String FALLBACK_CLUSTER_NAME = "fallback-" + CLUSTER_NAME;
   private static final String EDS_NAME = "eds-service-0";
   private static final String FALLBACK_EDS_NAME = "fallback-" + EDS_NAME;
+  private static final String AUTHORITY_NAME = "authority.example.com";
+  private static final String AUTHORITY_LDS_NAME =
+      "xdstp://" + AUTHORITY_NAME + "/envoy.config.listener.v3.Listener/listener1";
   private static final HttpConnectionManager MAIN_HTTP_CONNECTION_MANAGER =
       HttpConnectionManager.forRdsName(0, RDS_NAME, ImmutableList.of(
           new Filter.NamedFilterConfig("terminal-filter", RouterFilter.ROUTER_CONFIG)));
@@ -95,6 +99,7 @@ public class XdsClientFallbackTest {
   private ObjectPool<XdsClient> xdsClientPool;
   private XdsClient xdsClient;
   private boolean originalEnableXdsFallback;
+  private boolean originalEnableEndpointFallback;
   private final FakeClock fakeClock = new FakeClock();
   private final MetricRecorder metricRecorder = new MetricRecorder() {};
 
@@ -175,6 +180,8 @@ public void onAmbientError(Status error) {
   @Before
   public void setUp() throws XdsInitializationException {
     originalEnableXdsFallback = CommonBootstrapperTestUtils.setEnableXdsFallback(true);
+    originalEnableEndpointFallback =
+        CommonBootstrapperTestUtils.setEnableEndpointFallback(false);
     if (mainXdsServer == null) {
       throw new XdsInitializationException("Failed to create ControlPlaneRule for main TD server");
     }
@@ -194,6 +201,7 @@ public void cleanUp() {
       xdsClient = xdsClientPool.returnObject(xdsClient);
     }
     CommonBootstrapperTestUtils.setEnableXdsFallback(originalEnableXdsFallback);
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(originalEnableEndpointFallback);
   }
 
   private static void setAdsConfig(ControlPlaneRule controlPlane, String serverName) {
@@ -583,6 +591,106 @@ public void used_then_mainServerRestart_fallbackServerUp() {
     assertThat(getLrsServerInfo("localhost:" + fallbackServer.getServer().getPort())).isNull();
   }
 
+  /**
+   * gRFC A95: for an authority with fallback_on_reachability_only, falling back must happen when
+   * the primary server becomes unreachable, even though every subscribed resource is cached.
+   */
+  @Test
+  public void connect_then_mainServerDown_fallbackOnReachabilityOnly() throws Exception {
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(true);
+    verifyReachabilityOnlyFallback(true);
+  }
+
+  /**
+   * Without the knob, gRFC A71 behavior applies: cached resources suppress fallback entirely.
+   */
+  @Test
+  public void connect_then_mainServerDown_noReachabilityOnlyKnob_staysOnCache() throws Exception {
+    CommonBootstrapperTestUtils.setEnableEndpointFallback(true);
+    verifyReachabilityOnlyFallback(false);
+  }
+
+  private void verifyReachabilityOnlyFallback(boolean fallbackOnReachabilityOnly) throws Exception {
+    mainXdsServer.restartXdsServer();
+    fallbackServer.restartXdsServer();
+    // Serve the same xdstp resource from both servers, with distinguishable contents. The Listener
+    // name must be the xdstp resource name, since that is what the client matches against.
+    mainXdsServer.getService().setXdsConfig(ADS_TYPE_URL_LDS,
+        ImmutableMap.of(AUTHORITY_LDS_NAME,
+            ControlPlaneRule.buildClientListener(AUTHORITY_LDS_NAME, RDS_NAME)));
+    fallbackServer.getService().setXdsConfig(ADS_TYPE_URL_LDS,
+        ImmutableMap.of(AUTHORITY_LDS_NAME,
+            ControlPlaneRule.buildClientListener(AUTHORITY_LDS_NAME, FALLBACK_RDS_NAME)));
+
+    ExecutorService executor = Executors.newFixedThreadPool(1);
+    XdsTransportFactory xdsTransportFactory = new XdsTransportFactory() {
+      @Override
+      public XdsTransport create(Bootstrapper.ServerInfo serverInfo) {
+        ChannelCredentials channelCredentials =
+            (ChannelCredentials) serverInfo.implSpecificConfig();
+        return new GrpcXdsTransportFactory.GrpcXdsTransport(
+            Grpc.newChannelBuilder(serverInfo.target(), channelCredentials)
+                .executor(executor)
+                .build());
+      }
+    };
+    XdsClientImpl xdsClient = CommonBootstrapperTestUtils.createXdsClient(
+        new GrpcBootstrapperImpl().bootstrap(
+            authorityBootstrapOverride(fallbackOnReachabilityOnly)),
+        xdsTransportFactory, fakeClock, new ExponentialBackoffPolicy.Provider(),
+        MessagePrinter.INSTANCE, xdsClientMetricReporter);
+
+    xdsClient.watchXdsResource(
+        XdsListenerResource.getInstance(), AUTHORITY_LDS_NAME, ldsWatcher);
+
+    // Initial resource fetch from the main server; the resource is now cached.
+    verify(ldsWatcher, timeout(5000)).onResourceChanged(
+        StatusOr.fromValue(LdsUpdate.forApiListener(MAIN_HTTP_CONNECTION_MANAGER)));
+
+    mainXdsServer.getServer().shutdownNow();
+    // Sleep for the ADS stream disconnect to be processed and for the retry to fail. Between those
+    // two sleeps we need the fakeClock to progress by 1 second to restart the ADS stream.
+    for (int i = 0; i < 5; i++) {
+      // FakeClock is not thread-safe, and the retry scheduling is concurrent to this test thread
+      executor.submit(() -> fakeClock.forwardTime(1000, TimeUnit.MILLISECONDS)).get();
+      TimeUnit.SECONDS.sleep(1);
+    }
+
+    if (fallbackOnReachabilityOnly) {
+      // Falls back purely because the primary is unreachable, despite the resource being cached.
+      verify(ldsWatcher, timeout(5000)).onResourceChanged(
+          StatusOr.fromValue(LdsUpdate.forApiListener(FALLBACK_HTTP_CONNECTION_MANAGER)));
+    } else {
+      verify(ldsWatcher, never()).onResourceChanged(
+          StatusOr.fromValue(LdsUpdate.forApiListener(FALLBACK_HTTP_CONNECTION_MANAGER)));
+    }
+  }
+
+  private Map<String, ?> authorityBootstrapOverride(boolean fallbackOnReachabilityOnly) {
+    ImmutableList<?> servers = ImmutableList.of(
+        ImmutableMap.of(
+            "server_uri", "localhost:" + mainXdsServer.getServer().getPort(),
+            "channel_creds", Collections.singletonList(ImmutableMap.of("type", "insecure")),
+            "server_features", Collections.singletonList("xds_v3")
+        ),
+        ImmutableMap.of(
+            "server_uri", "localhost:" + fallbackServer.getServer().getPort(),
+            "channel_creds", Collections.singletonList(ImmutableMap.of("type", "insecure")),
+            "server_features", Collections.singletonList("xds_v3")
+        ));
+    return ImmutableMap.of(
+        "node", ImmutableMap.of(
+            "id", UUID.randomUUID().toString(),
+            "cluster", CLUSTER_NAME),
+        "xds_servers", servers,
+        "authorities", ImmutableMap.of(
+            AUTHORITY_NAME, ImmutableMap.of(
+                "xds_servers", servers,
+                "fallback_on_reachability_only", fallbackOnReachabilityOnly)),
+        "fallback-policy", "fallback"
+    );
+  }
+
   private Map<String, ?> defaultBootstrapOverride() {
     return ImmutableMap.of(
         "node", ImmutableMap.of(

xds/src/test/java/io/grpc/xds/client/CommonBootstrapperTestUtils.java

@@ -166,6 +166,13 @@ public static boolean setEnableXdsFallback(boolean target) {
     return oldValue;
   }
 
+  /** Toggles the gRFC A95 endpoint fallback env var guard; returns the previous value. */
+  public static boolean setEnableEndpointFallback(boolean target) {
+    boolean oldValue = BootstrapperImpl.enableEndpointFallback;
+    BootstrapperImpl.enableEndpointFallback = target;
+    return oldValue;
+  }
+
   public static XdsClientImpl createXdsClient(List<String> serverUris,
                                               XdsTransportFactory xdsTransportFactory,
                                               FakeClock fakeClock,