Read in absence of viable writer

Previously driver did not allow reads and writes when received routing
table did not contains both routers, readers and writers. This was
inconsistent with Causal Cluster which allows reads when leader is absent.
Leader might be unavailable for a long time (when there is a DC failure, etc.)
so it makes sense to allow clients to perform read activity.

This commit makes driver accept routing table with no writers and allow
clients to perform read operations when writers are not available. It
might be problematic when there is a cluster partition and one partition
contains majority. For this case special care must be taken so driver
does not get stuck talking only to the smaller partition which only
knows about itself. This is done on best effort basis - driver tries
to contact seed router if previously accepted routing table did not
contain writes.

Author: lutovich

driver/src/main/java/org/neo4j/driver/internal/NetworkSession.java

@@ -18,8 +18,6 @@
  */
 package org.neo4j.driver.internal;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -378,28 +376,4 @@ private void closeCurrentConnection( boolean sync )
             logger.debug( "Released connection " + connection.hashCode() );
         }
     }
-
-    private static List<Throwable> recordError( Throwable error, List<Throwable> errors )
-    {
-        if ( errors == null )
-        {
-            errors = new ArrayList<>();
-        }
-        errors.add( error );
-        return errors;
-    }
-
-    private static void addSuppressed( Throwable error, List<Throwable> suppressedErrors )
-    {
-        if ( suppressedErrors != null )
-        {
-            for ( Throwable suppressedError : suppressedErrors )
-            {
-                if ( error != suppressedError )
-                {
-                    error.addSuppressed( suppressedError );
-                }
-            }
-        }
-    }
 }

driver/src/main/java/org/neo4j/driver/internal/cluster/ClusterRoutingTable.java

@@ -25,6 +25,7 @@
 
 import org.neo4j.driver.internal.net.BoltServerAddress;
 import org.neo4j.driver.internal.util.Clock;
+import org.neo4j.driver.v1.AccessMode;
 
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -34,7 +35,7 @@ public class ClusterRoutingTable implements RoutingTable
     private static final int MIN_ROUTERS = 1;
 
     private final Clock clock;
-    private long expirationTimeout;
+    private volatile long expirationTimeout;
     private final RoundRobinAddressSet readers;
     private final RoundRobinAddressSet writers;
     private final RoundRobinAddressSet routers;
@@ -56,12 +57,12 @@ private ClusterRoutingTable( Clock clock )
     }
 
     @Override
-    public boolean isStale()
+    public boolean isStaleFor( AccessMode mode )
     {
-        return expirationTimeout < clock.millis() || // the expiration timeout has been reached
-               routers.size() <= MIN_ROUTERS || // we need to discover more routing servers
-               readers.size() == 0 || // we need to discover more read servers
-               writers.size() == 0; // we need to discover more write servers
+        return expirationTimeout < clock.millis() ||
+               routers.size() <= MIN_ROUTERS ||
+               mode == AccessMode.READ && readers.size() == 0 ||
+               mode == AccessMode.WRITE && writers.size() == 0;
     }
 
     @Override
@@ -115,7 +116,7 @@ public void removeWriter( BoltServerAddress toRemove )
 
 
     @Override
-    public String toString()
+    public synchronized String toString()
     {
         return format( "Ttl %s, currentTime %s, routers %s, writers %s, readers %s",
                 expirationTimeout, clock.millis(), routers, writers, readers );

driver/src/main/java/org/neo4j/driver/internal/cluster/LoadBalancer.java

@@ -29,8 +29,8 @@
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.Logger;
 import org.neo4j.driver.v1.Logging;
-import org.neo4j.driver.v1.exceptions.ProtocolException;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
+import org.neo4j.driver.v1.exceptions.SessionExpiredException;
 
 public class LoadBalancer implements ConnectionProvider, RoutingErrorHandler, AutoCloseable
 {
@@ -61,15 +61,14 @@ private LoadBalancer( BoltServerAddress initialRouter, RoutingSettings settings,
         this.rediscovery = rediscovery;
         this.log = log;
 
-        // initialize the routing table
-        ensureRouting();
+        refreshRoutingTable();
     }
 
     @Override
     public PooledConnection acquireConnection( AccessMode mode )
     {
         RoundRobinAddressSet addressSet = addressSetFor( mode );
-        PooledConnection connection = acquireConnection( addressSet );
+        PooledConnection connection = acquireConnection( mode, addressSet );
         return new RoutingPooledConnection( connection, this, mode );
     }
 
@@ -91,26 +90,23 @@ public void close() throws Exception
         connections.close();
     }
 
-    private PooledConnection acquireConnection( RoundRobinAddressSet servers ) throws ServiceUnavailableException
+    private PooledConnection acquireConnection( AccessMode mode, RoundRobinAddressSet servers )
     {
-        for ( ; ; )
+        ensureRouting( mode );
+        for ( BoltServerAddress address; (address = servers.next()) != null; )
         {
-            // refresh the routing table if needed
-            ensureRouting();
-            for ( BoltServerAddress address; (address = servers.next()) != null; )
+            try
             {
-                try
-                {
-                    return connections.acquire( address );
-                }
-                catch ( ServiceUnavailableException e )
-                {
-                    log.error( "Failed to obtain a connection towards address " + address, e );
-                    forget( address );
-                }
+                return connections.acquire( address );
+            }
+            catch ( ServiceUnavailableException e )
+            {
+                log.error( "Failed to obtain a connection towards address " + address, e );
+                forget( address );
             }
-            // if we get here, we failed to connect to any server, so we will rebuild the routing table
         }
+        throw new SessionExpiredException(
+                "Failed to obtain connection towards " + mode + " server. Known routing table is: " + routingTable );
     }
 
     private synchronized void forget( BoltServerAddress address )
@@ -121,23 +117,28 @@ private synchronized void forget( BoltServerAddress address )
         connections.purge( address );
     }
 
-    synchronized void ensureRouting() throws ServiceUnavailableException, ProtocolException
+    synchronized void ensureRouting( AccessMode mode )
     {
-        if ( routingTable.isStale() )
+        if ( routingTable.isStaleFor( mode ) )
         {
-            log.info( "Routing information is stale. %s", routingTable );
+            refreshRoutingTable();
+        }
+    }
 
-            // get a new routing table
-            ClusterComposition cluster = rediscovery.lookupClusterComposition( connections, routingTable );
-            Set<BoltServerAddress> removed = routingTable.update( cluster );
-            // purge connections to removed addresses
-            for ( BoltServerAddress address : removed )
-            {
-                connections.purge( address );
-            }
+    synchronized void refreshRoutingTable()
+    {
+        log.info( "Routing information is stale. %s", routingTable );
 
-            log.info( "Refreshed routing information. %s", routingTable );
+        // get a new routing table
+        ClusterComposition cluster = rediscovery.lookupClusterComposition( routingTable, connections );
+        Set<BoltServerAddress> removed = routingTable.update( cluster );
+        // purge connections to removed addresses
+        for ( BoltServerAddress address : removed )
+        {
+            connections.purge( address );
         }
+
+        log.info( "Refreshed routing information. %s", routingTable );
     }
 
     private RoundRobinAddressSet addressSetFor( AccessMode mode )

driver/src/main/java/org/neo4j/driver/internal/cluster/Rediscovery.java

@@ -18,6 +18,7 @@
  */
 package org.neo4j.driver.internal.cluster;
 
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -42,6 +43,8 @@ public class Rediscovery
     private final ClusterCompositionProvider provider;
     private final HostNameResolver hostNameResolver;
 
+    private boolean useInitialRouter;
+
     public Rediscovery( BoltServerAddress initialRouter, RoutingSettings settings, Clock clock, Logger logger,
             ClusterCompositionProvider provider, HostNameResolver hostNameResolver )
     {
@@ -53,9 +56,15 @@ public Rediscovery( BoltServerAddress initialRouter, RoutingSettings settings, C
         this.hostNameResolver = hostNameResolver;
     }
 
-    // Given the current routing table and connection pool, use the connection composition provider to fetch a new
-    // cluster composition, which would be used to update the routing table and connection pool
-    public ClusterComposition lookupClusterComposition( ConnectionPool connections, RoutingTable routingTable )
+    /**
+     * Given the current routing table and connection pool, use the connection composition provider to fetch a new
+     * cluster composition, which would be used to update the routing table and connection pool.
+     *
+     * @param routingTable current routing table.
+     * @param connections connection pool.
+     * @return new cluster composition.
+     */
+    public ClusterComposition lookupClusterComposition( RoutingTable routingTable, ConnectionPool connections )
     {
         int failures = 0;
 
@@ -65,7 +74,7 @@ public ClusterComposition lookupClusterComposition( ConnectionPool connections,
             sleep( waitTime );
             start = clock.millis();
 
-            ClusterComposition composition = lookupClusterCompositionOnKnownRouters( connections, routingTable );
+            ClusterComposition composition = lookup( routingTable, connections );
             if ( composition != null )
             {
                 return composition;
@@ -78,11 +87,56 @@ public ClusterComposition lookupClusterComposition( ConnectionPool connections,
         }
     }
 
-    private ClusterComposition lookupClusterCompositionOnKnownRouters( ConnectionPool connections,
-            RoutingTable routingTable )
+    private ClusterComposition lookup( RoutingTable routingTable, ConnectionPool connections )
+    {
+        ClusterComposition composition;
+
+        if ( useInitialRouter )
+        {
+            composition = lookupOnInitialRouterThenOnKnownRouters( routingTable, connections );
+            useInitialRouter = false;
+        }
+        else
+        {
+            composition = lookupOnKnownRoutersThenOnInitialRouter( routingTable, connections );
+        }
+
+        if ( composition != null && !composition.hasWriters() )
+        {
+            useInitialRouter = true;
+        }
+
+        return composition;
+    }
+
+    private ClusterComposition lookupOnKnownRoutersThenOnInitialRouter( RoutingTable routingTable,
+            ConnectionPool connections )
+    {
+        Set<BoltServerAddress> seenServers = new HashSet<>();
+        ClusterComposition composition = lookupOnKnownRouters( routingTable, connections, seenServers );
+        if ( composition == null )
+        {
+            return lookupOnInitialRouter( routingTable, connections, seenServers );
+        }
+        return composition;
+    }
+
+    private ClusterComposition lookupOnInitialRouterThenOnKnownRouters( RoutingTable routingTable,
+            ConnectionPool connections )
+    {
+        Set<BoltServerAddress> seenServers = Collections.emptySet();
+        ClusterComposition composition = lookupOnInitialRouter( routingTable, connections, seenServers );
+        if ( composition == null )
+        {
+            return lookupOnKnownRouters( routingTable, connections, new HashSet<BoltServerAddress>() );
+        }
+        return composition;
+    }
+
+    private ClusterComposition lookupOnKnownRouters( RoutingTable routingTable, ConnectionPool connections,
+            Set<BoltServerAddress> seenServers )
     {
         int size = routingTable.routerSize();
-        Set<BoltServerAddress> triedServers = new HashSet<>();
         for ( int i = 0; i < size; i++ )
         {
             BoltServerAddress address = routingTable.nextRouter();
@@ -91,22 +145,28 @@ private ClusterComposition lookupClusterCompositionOnKnownRouters( ConnectionPoo
                 break;
             }
 
-            ClusterComposition composition = lookupClusterCompositionOnRouter( address, connections, routingTable );
+            ClusterComposition composition = lookupOnRouter( address, routingTable, connections );
             if ( composition != null )
             {
                 return composition;
             }
             else
             {
-                triedServers.add( address );
+                seenServers.add( address );
             }
         }
 
+        return null;
+    }
+
+    private ClusterComposition lookupOnInitialRouter( RoutingTable routingTable,
+            ConnectionPool connections, Set<BoltServerAddress> triedServers )
+    {
         Set<BoltServerAddress> ips = hostNameResolver.resolve( initialRouter );
         ips.removeAll( triedServers );
         for ( BoltServerAddress address : ips )
         {
-            ClusterComposition composition = lookupClusterCompositionOnRouter( address, connections, routingTable );
+            ClusterComposition composition = lookupOnRouter( address, routingTable, connections );
             if ( composition != null )
             {
                 return composition;
@@ -116,8 +176,8 @@ private ClusterComposition lookupClusterCompositionOnKnownRouters( ConnectionPoo
         return null;
     }
 
-    private ClusterComposition lookupClusterCompositionOnRouter( BoltServerAddress routerAddress,
-            ConnectionPool connections, RoutingTable routingTable )
+    private ClusterComposition lookupOnRouter( BoltServerAddress routerAddress, RoutingTable routingTable,
+            ConnectionPool connections )
     {
         ClusterCompositionResponse response;
         try ( Connection connection = connections.acquire( routerAddress ) )
@@ -139,11 +199,7 @@ private ClusterComposition lookupClusterCompositionOnRouter( BoltServerAddress r
 
         ClusterComposition cluster = response.clusterComposition();
         logger.info( "Got cluster composition %s", cluster );
-        if ( cluster.hasWriters() )
-        {
-            return cluster;
-        }
-        return null;
+        return cluster;
     }
 
     private void sleep( long millis )

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingTable.java

@@ -21,10 +21,11 @@
 import java.util.Set;
 
 import org.neo4j.driver.internal.net.BoltServerAddress;
+import org.neo4j.driver.v1.AccessMode;
 
 public interface RoutingTable
 {
-    boolean isStale();
+    boolean isStaleFor( AccessMode mode );
 
     Set<BoltServerAddress> update( ClusterComposition cluster );
 

driver/src/main/java/org/neo4j/driver/internal/net/pooling/SocketConnectionPool.java

@@ -69,7 +69,7 @@ public SocketConnectionPool( PoolSettings poolSettings, Connector connector, Clo
     }
 
     @Override
-    public PooledConnection acquire( final BoltServerAddress address )
+    public PooledConnection acquire( BoltServerAddress address )
     {
         assertNotClosed();
         BlockingPooledConnectionQueue connectionQueue = pool( address );