Fix Android Connectivity Monitor (v2)

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java

@@ -174,6 +174,11 @@ public void run() {
   /** The time a stream stays open after it is marked idle. */
   private static final long IDLE_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(1);
 
+  /**
+   * Maximum backoff time for reconnecting when we know the connection is failed on the client-side.
+   */
+  private static final long DNS_FAILURE_BACKOFF_MAX_DELAY_MS = TimeUnit.SECONDS.toMillis(10);
+
   @Nullable private DelayedTask idleTimer;
 
   private final FirestoreChannel firestoreChannel;
@@ -290,18 +295,22 @@ private void close(State finalState, Status status) {
     if (code == Code.OK) {
       // If this is an intentional close ensure we don't delay our next connection attempt.
       backoff.reset();
-
     } else if (code == Code.RESOURCE_EXHAUSTED) {
       Logger.debug(
           getClass().getSimpleName(),
           "(%x) Using maximum backoff delay to prevent overloading the backend.",
           System.identityHashCode(this));
       backoff.resetToMax();
-
     } else if (code == Code.UNAUTHENTICATED) {
       // "unauthenticated" error means the token was rejected. Try force refreshing it in case it
       // just expired.
       firestoreChannel.invalidateToken();
+    } else if (code == Code.UNAVAILABLE) {
+      // This exception is thrown when the gRPC connection fails on the client side, To shorten
+      // reconnect time, we can use a shorter max delay when reconnecting.
+      if (status.getCause() instanceof java.net.UnknownHostException) {
+        backoff.setTemporaryMaxDelay(DNS_FAILURE_BACKOFF_MAX_DELAY_MS);
+      }
     }
 
     if (finalState != State.Error) {

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/GrpcCallProvider.java

@@ -23,13 +23,16 @@
 import com.google.android.gms.tasks.Tasks;
 import com.google.firebase.firestore.core.DatabaseInfo;
 import com.google.firebase.firestore.util.AsyncQueue;
+import com.google.firebase.firestore.util.AsyncQueue.DelayedTask;
+import com.google.firebase.firestore.util.AsyncQueue.TimerId;
 import com.google.firebase.firestore.util.Executors;
 import com.google.firebase.firestore.util.Logger;
 import com.google.firebase.firestore.util.Supplier;
 import com.google.firestore.v1.FirestoreGrpc;
 import io.grpc.CallCredentials;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
+import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.MethodDescriptor;
@@ -45,11 +48,23 @@ public class GrpcCallProvider {
 
   private static Supplier<ManagedChannelBuilder<?>> overrideChannelBuilderSupplier;
 
-  private final Task<ManagedChannel> channelTask;
+  private Task<ManagedChannel> channelTask;
   private final AsyncQueue asyncQueue;
 
   private CallOptions callOptions;
 
+  // A timer that elapses after CONNECTIVTY_ATTEMPT_TIMEOUT_MS, at which point we restart the
+  // channel and try connecting again.
+  private DelayedTask connectivityAttemptTimer;
+
+  // This timeout is used when attempting to establish a connection in gRPC. If a connection attempt
+  // does not succeed in time, we restart the channel, rather than having it hang indefinitely.
+  private static final int CONNECTIVITY_ATTEMPT_TIMEOUT_MS = 15 * 1000;
+
+  private final Context context;
+  private final DatabaseInfo databaseInfo;
+  private final CallCredentials firestoreHeaders;
+
   /**
    * Helper function to globally override the channel that RPCs use. Useful for testing when you
    * want to bypass SSL certificate checking.
@@ -69,24 +84,11 @@ public static void overrideChannelBuilder(
       DatabaseInfo databaseInfo,
       CallCredentials firestoreHeaders) {
     this.asyncQueue = asyncQueue;
+    this.context = context;
+    this.databaseInfo = databaseInfo;
+    this.firestoreHeaders = firestoreHeaders;
 
-    // We execute network initialization on a separate thread to not block operations that depend on
-    // the AsyncQueue.
-    this.channelTask =
-        Tasks.call(
-            Executors.BACKGROUND_EXECUTOR,
-            () -> {
-              ManagedChannel channel = initChannel(context, databaseInfo);
-              FirestoreGrpc.FirestoreStub firestoreStub =
-                  FirestoreGrpc.newStub(channel)
-                      .withCallCredentials(firestoreHeaders)
-                      // Ensure all callbacks are issued on the worker queue. If this call is
-                      // removed, all calls need to be audited to make sure they are executed on the
-                      // right thread.
-                      .withExecutor(asyncQueue.getExecutor());
-              callOptions = firestoreStub.getCallOptions();
-              return channel;
-            });
+    initChannelTask();
   }
 
   /** Sets up the SSL provider and configures the gRPC channel. */
@@ -198,4 +200,71 @@ void shutdown() {
       Thread.currentThread().interrupt();
     }
   }
+
+  /**
+   * Cancels the connectivityStateTimer if the new state indicates gRPC is online. Starts the timer
+   * if the new state indicates gRPC is connecting.
+   */
+  private void onConnectivityStateChange(ManagedChannel channel) {
+    ConnectivityState newState = channel.getState(false);
+    // Check that the new state is online, then cancel timer.
+    if (newState == ConnectivityState.READY) {
+      clearConnectivityTimer();
+    } else if (newState == ConnectivityState.CONNECTING) {
+      connectivityAttemptTimer =
+          asyncQueue.enqueueAfterDelay(
+              TimerId.CONNECTIVITY_ATTEMPT_TIMER,
+              CONNECTIVITY_ATTEMPT_TIMEOUT_MS,
+              () -> {
+                clearConnectivityTimer();
+                resetChannel(channel);
+              });
+    }
+    // Re-listen for next state change.
+    channel.notifyWhenStateChanged(
+        newState, () -> asyncQueue.enqueueAndForget(() -> onConnectivityStateChange(channel)));
+  }
+
+  /**
+   * Shuts down and re-initializes the channel.
+   *
+   * <p>This is used when the connectivity attempt timer elapses and we need to reset the gRPC
+   * channel to re-establish connectivity.
+   */
+  private void resetChannel(ManagedChannel channel) {
+    asyncQueue.enqueueAndForget(
+        () -> {
+          channel.shutdownNow();
+          initChannelTask();
+        });
+  }
+
+  private void initChannelTask() {
+    // We execute network initialization on a separate thread to not block operations that depend on
+    // the AsyncQueue.
+    this.channelTask =
+        Tasks.call(
+            Executors.BACKGROUND_EXECUTOR,
+            () -> {
+              ManagedChannel channel = initChannel(context, databaseInfo);
+              onConnectivityStateChange(channel);
+              FirestoreGrpc.FirestoreStub firestoreStub =
+                  FirestoreGrpc.newStub(channel)
+                      .withCallCredentials(firestoreHeaders)
+                      // Ensure all callbacks are issued on the worker queue. If this call is
+                      // removed, all calls need to be audited to make sure they are executed on the
+                      // right thread.
+                      .withExecutor(asyncQueue.getExecutor());
+              callOptions = firestoreStub.getCallOptions();
+              return channel;
+            });
+  }
+
+  /** Clears the connectivity timer if it exists. */
+  private void clearConnectivityTimer() {
+    if (connectivityAttemptTimer != null) {
+      connectivityAttemptTimer.cancel();
+      connectivityAttemptTimer = null;
+    }
+  }
 }

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/OnlineStateTracker.java

@@ -143,8 +143,8 @@ void handleWatchStreamFailure(Status status) {
         logClientOfflineWarningIfNecessary(
             String.format(
                 Locale.ENGLISH,
-                "Connection failed %d times. Most recent error: %s",
-                MAX_WATCH_STREAM_FAILURES,
+                "Backend didn't respond within %d seconds. Most recent error: %s\n",
+                ONLINE_STATE_TIMEOUT_MS / 1000,
                 status));
         setAndBroadcastState(OnlineState.OFFLINE);
       }

firebase-firestore/src/main/java/com/google/firebase/firestore/util/AsyncQueue.java

@@ -74,7 +74,12 @@ public enum TimerId {
      * A timer used to retry transactions. Since there can be multiple concurrent transactions,
      * multiple of these may be in the queue at a given time.
      */
-    RETRY_TRANSACTION
+    RETRY_TRANSACTION,
+    /**
+     * A timer used to monitor when a connection attempt in gRPC is unsuccessful and retry
+     * accordingly.
+     */
+    CONNECTIVITY_ATTEMPT_TIMER
   }
 
   /**

firebase-firestore/src/main/java/com/google/firebase/firestore/util/ExponentialBackoff.java

@@ -36,8 +36,8 @@ public class ExponentialBackoff {
   private final TimerId timerId;
   private final long initialDelayMs;
   private final double backoffFactor;
-  private final long maxDelayMs;
 
+  private long maxDelayMs;
   private long currentBaseMs;
   private long lastAttemptTime;
   private DelayedTask timerTask;
@@ -103,6 +103,16 @@ public void resetToMax() {
     currentBaseMs = maxDelayMs;
   }
 
+  /**
+   * Set the backoff's maximum delay for only the next call to backoffAndRun, after which the delay
+   * will be reset to maxDelayMs.
+   *
+   * @param newMax The temporary maximum delay to set.
+   */
+  public void setTemporaryMaxDelay(long newMax) {
+    maxDelayMs = newMax;
+  }
+
   /**
    * Waits for currentDelayMs, increases the delay and runs the specified task. If there was a
    * pending backoff task waiting to run already, it will be canceled.
@@ -151,6 +161,9 @@ public void backoffAndRun(Runnable task) {
     } else if (currentBaseMs > maxDelayMs) {
       currentBaseMs = maxDelayMs;
     }
+
+    // Reset max delay to the default.
+    maxDelayMs = DEFAULT_BACKOFF_MAX_DELAY_MS;
   }
 
   public void cancel() {