Repository: ably/ably-java
Issue: Long-lived connections are immediately transitioned to `SUSPENDED` after disconnection

Relates to spec:

- `DF1a`
- `RTN14d`
- `RTN15g`

The default `connectionStateTtl` value (how long the connection is persisted on the Ably servers) is 120 seconds. This means that after 120 seconds of being `DISCONNECTED`, the connection state should transition to `SUSPENDED`.

We maintain a timer (`ConnectionManager.suspendTime`) in `ably-java` that represents the time at which suspension should occur. This is checked whenever:

- A `CONNECTING` state times out
- The transport is unavailable

If the time is not exceeded, the state is transitioned to `DISCONNECTED`. If it is exceeded, then the state becomes `SUSPENDED`.

The `suspendTime` is reset (to `now + connetionStateTtl`) only:

- On `ConnectionManager` initialisation
- On successful connection
- When a `DISCONNECTED` state is enacted, the previous state was `CONNECTED` and immediate retries are not suppressed.

If the only thing a connection has done is successfully connect... and transport becomes unavailable (via connection rebalancing, internet issues etc)... and we've been connected for >`connectionStateTtl`... then `ConnectionManager.onTransportUnavailable` will check the timer, see that it has passed, and immediately suspend the connection.

The correct behaviour here is to enter a `DISCONNECTED` state and immediately retry connections, only suspending at `disconnectTime + connectionStateTtl`.

Changed files:
lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java, lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

diff --git a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
index 09ab6505..f326382c 100644
--- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
+++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
@@ -309,10 +309,13 @@ public class ConnectionManager implements ConnectListener {
         void enact(StateIndication stateIndication, ConnectionStateChange change) {
             super.enact(stateIndication, change);
             clearTransport();
+
+            // If we were connected, immediately retry
             if(change.previous == ConnectionState.connected) {
                 setSuspendTime();
-                /* we were connected, so retry immediately */
-                if(!suppressRetry) {
+
+                if (!suppressRetry) {
+                    Log.v(TAG, "Was previously connected, retrying immediately");
                     requestState(ConnectionState.connecting);
                 }
             }
@@ -1253,7 +1256,6 @@ public class ConnectionManager implements ConnectListener {
             return;
         }
         /* indicated connected currentState */
-        setSuspendTime();
         final StateIndication stateIndication = new StateIndication(ConnectionState.connected, error, null, null,
             reattachOnResumeFailure);
         requestState(stateIndication);
@@ -1459,12 +1461,18 @@ public class ConnectionManager implements ConnectListener {
 
     @Override
     public synchronized void onTransportUnavailable(ITransport transport, ErrorInfo reason) {
+        Log.v(TAG, "onTransportUnavailable()");
         if (this.transport != transport) {
             /* This is from a transport that we have already abandoned. */
             Log.v(TAG, "onTransportUnavailable: ignoring disconnection event from superseded transport");
             return;
         }
 
+        // If we're currently connected, start the suspend timer
+        if (currentState.state == ConnectionState.connected) {
+            setSuspendTime();
+        }
+
         /* if this is a failure of a pending connection attempt, decide whether or not to attempt a fallback host */
         StateIndication fallbackAttempt = checkFallback(reason);
         if(fallbackAttempt != null) {
diff --git a/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java b/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java
index 602b9d4f..2182f537 100644
--- a/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java
+++ b/lib/src/test/java/io/ably/lib/test/realtime/ConnectionManagerTest.java
@@ -20,14 +20,17 @@ import io.ably.lib.test.util.MockWebsocketFactory;
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.transport.Defaults;
 import io.ably.lib.transport.Hosts;
+import io.ably.lib.transport.ITransport;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
+import io.ably.lib.types.ErrorInfo;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
 import org.mockito.Mockito;
 
+import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -731,4 +734,81 @@ public class ConnectionManagerTest extends ParameterizedTest {
             assertEquals("Suspended channel histories do not match", suspendedChannelHistory, expectedSuspendedChannelHistory);
         }
     }
+
+    /**
+     * <p>
+     * Verifies that the {@code ConnectionManager} enters the disconnected state and sets the suspend timer
+     * upon unavailable transport.
+     * </p>
+     * <p>
+     * Spec: RTN15g
+     * </p>
+     */
+    @Test
+    public void connection_manager_enters_disconnected_state_on_transport_failure() throws AblyException, NoSuchFieldException, IllegalAccessException {
+        ClientOptions opts = createOptions(testVars.keys[0].keyStr);
+        try(AblyRealtime ably = new AblyRealtime(opts)) {
+            ConnectionManager connectionManager = ably.connection.connectionManager;
+            connectionManager.connect();
+            new Helpers.ConnectionManagerWaiter(ably.connection.connectionManager).waitFor(ConnectionState.connected);
+
+            // Here, we "fake" being online for 2 minutes - the suspendTime is set by onConnected and the default is 2 minutes
+            Field suspendTimeField = connectionManager.getClass().getDeclaredField("suspendTime");
+            suspendTimeField.setAccessible(true);
+            suspendTimeField.set(connectionManager, System.currentTimeMillis() - 10);
+
+            // We also have to grab the "real" transport to pass the superseded test
+            Field transportField = connectionManager.getClass().getDeclaredField("transport");
+            transportField.setAccessible(true);
+
+            connectionManager.onTransportUnavailable((ITransport) transportField.get(connectionManager), new ErrorInfo());
+            new Helpers.ConnectionManagerWaiter(connectionManager).waitFor(ConnectionState.disconnected);
+
+            assertTrue((long) suspendTimeField.get(connectionManager) >= System.currentTimeMillis());
+
+            connectionManager.close();
+        }
+    }
+
+    /**
+     * <p>
+     * Verifies that the {@code ConnectionManager} enters the suspended state if the transport is unavailable and the
+     * timer has been exceeded.
+     * </p>
+     * <p>
+     * Spec: RTN15g, RTN14d
+     * </p>
+     */
+    @Test
+    public void connection_manager_enters_suspended_state_on_transport_failure_after_already_being_disconnected_for_2_minutes() throws AblyException, NoSuchFieldException, IllegalAccessException, ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException {
+        ClientOptions opts = createOptions(testVars.keys[0].keyStr);
+        try(AblyRealtime ably = new AblyRealtime(opts)) {
+            ConnectionManager connectionManager = ably.connection.connectionManager;
+            connectionManager.connect();
+            new Helpers.ConnectionManagerWaiter(ably.connection.connectionManager).waitFor(ConnectionState.connected);
+
+            // Here, we "fake" being disconnected beyond the suspend timer
+            Class<?> connectionManagerClass = Class.forName("io.ably.lib.transport.ConnectionManager");
+            Class<?> disconnectedState = Class.forName("io.ably.lib.transport.ConnectionManager$Disconnected");
+            Constructor<?> disconnectedStateCtor = disconnectedState.getDeclaredConstructor(connectionManagerClass);
+            disconnectedStateCtor.setAccessible(true);
+            Field connectionStateField = connectionManager.getClass().getDeclaredField("currentState");
+            connectionStateField.setAccessible(true);
+            connectionStateField.set(connectionManager, disconnectedStateCtor.newInstance(connectionManager));
+
+            Field suspendTimeField = connectionManager.getClass().getDeclaredField("suspendTime");
+            suspendTimeField.setAccessible(true);
+            suspendTimeField.set(connectionManager, System.currentTimeMillis() - 5000);
+
+            // We also have to grab the "real" transport to pass the superseded test
+            Field transportField = connectionManager.getClass().getDeclaredField("transport");
+            transportField.setAccessible(true);
+
+            connectionManager.onTransportUnavailable((ITransport) transportField.get(connectionManager), new ErrorInfo());
+
+            new Helpers.ConnectionManagerWaiter(connectionManager).waitFor(ConnectionState.suspended);
+
+            connectionManager.close();
+        }
+    }
 }