Configurable retries for declarative transactions

Added ability to retry transactions executed via `Session#readTransaction()`
and `Session#writeTransaction()` methods. Given unit of work will be retried
configured number of times with configured delay if it experiences
`ServiceUnavailableException` or `SessionExpiredException`. This means it'll
be retried when a network error happens of cluster member becomes unavailable.

Author: lutovich

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

@@ -28,6 +28,10 @@
 import org.neo4j.driver.internal.net.SocketConnector;
 import org.neo4j.driver.internal.net.pooling.PoolSettings;
 import org.neo4j.driver.internal.net.pooling.SocketConnectionPool;
+import org.neo4j.driver.internal.retry.RetryDecision;
+import org.neo4j.driver.internal.retry.RetryLogic;
+import org.neo4j.driver.internal.retry.RetrySettings;
+import org.neo4j.driver.internal.retry.RetryWithDelay;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
@@ -47,15 +51,18 @@
 
 public class DriverFactory
 {
-    public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings routingSettings, Config config )
+    public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings routingSettings,
+            RetrySettings retrySettings, Config config )
     {
         BoltServerAddress address = BoltServerAddress.from( uri );
         SecurityPlan securityPlan = createSecurityPlan( address, config );
         ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, config );
+        RetryLogic<RetryDecision> retryLogic = RetryWithDelay.create( retrySettings, createClock() );
 
         try
         {
-            return createDriver( address, uri.getScheme(), connectionPool, config, routingSettings, securityPlan );
+            return createDriver( address, uri.getScheme(), connectionPool, config, routingSettings, securityPlan,
+                    retryLogic );
         }
         catch ( Throwable driverError )
         {
@@ -73,14 +80,15 @@ public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings r
     }
 
     private Driver createDriver( BoltServerAddress address, String scheme, ConnectionPool connectionPool,
-            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan )
+            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan,
+            RetryLogic<RetryDecision> retryLogic )
     {
         switch ( scheme.toLowerCase() )
         {
         case "bolt":
-            return createDirectDriver( address, connectionPool, config, securityPlan );
+            return createDirectDriver( address, connectionPool, config, securityPlan, retryLogic );
         case "bolt+routing":
-            return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan );
+            return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan, retryLogic );
         default:
             throw new ClientException( format( "Unsupported URI scheme: %s", scheme ) );
         }
@@ -92,10 +100,10 @@ private Driver createDriver( BoltServerAddress address, String scheme, Connectio
      * <b>This method is protected only for testing</b>
      */
     protected Driver createDirectDriver( BoltServerAddress address, ConnectionPool connectionPool, Config config,
-            SecurityPlan securityPlan )
+            SecurityPlan securityPlan, RetryLogic<RetryDecision> retryLogic )
     {
         ConnectionProvider connectionProvider = new DirectConnectionProvider( address, connectionPool );
-        SessionFactory sessionFactory = createSessionFactory( connectionProvider, config );
+        SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
         return createDriver( config, securityPlan, sessionFactory );
     }
 
@@ -105,14 +113,15 @@ protected Driver createDirectDriver( BoltServerAddress address, ConnectionPool c
      * <b>This method is protected only for testing</b>
      */
     protected Driver createRoutingDriver( BoltServerAddress address, ConnectionPool connectionPool,
-            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan )
+            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan,
+            RetryLogic<RetryDecision> retryLogic )
     {
         if ( !securityPlan.isRoutingCompatible() )
         {
             throw new IllegalArgumentException( "The chosen security plan is not compatible with a routing driver" );
         }
         ConnectionProvider connectionProvider = createLoadBalancer( address, connectionPool, config, routingSettings );
-        SessionFactory sessionFactory = createSessionFactory( connectionProvider, config );
+        SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
         return createDriver( config, securityPlan, sessionFactory );
     }
 
@@ -180,9 +189,10 @@ protected Connector createConnector( ConnectionSettings connectionSettings, Secu
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected SessionFactory createSessionFactory( ConnectionProvider connectionProvider, Config config )
+    protected SessionFactory createSessionFactory( ConnectionProvider connectionProvider,
+            RetryLogic<RetryDecision> retryLogic, Config config )
     {
-        return new SessionFactoryImpl( connectionProvider, config, config.logging() );
+        return new SessionFactoryImpl( connectionProvider, retryLogic, config );
     }
 
     private static SecurityPlan createSecurityPlan( BoltServerAddress address, Config config )

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

@@ -18,6 +18,8 @@
  */
 package org.neo4j.driver.internal;
 
+import org.neo4j.driver.internal.retry.RetryDecision;
+import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.Logging;
@@ -28,9 +30,10 @@ class LeakLoggingNetworkSession extends NetworkSession
 {
     private final String stackTrace;
 
-    LeakLoggingNetworkSession( ConnectionProvider connectionProvider, AccessMode mode, Logging logging )
+    LeakLoggingNetworkSession( ConnectionProvider connectionProvider, AccessMode mode,
+            RetryLogic<RetryDecision> retryLogic, Logging logging )
     {
-        super( connectionProvider, mode, logging );
+        super( connectionProvider, mode, retryLogic, logging );
         this.stackTrace = captureStackTrace();
     }
 

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

@@ -18,9 +18,13 @@
  */
 package org.neo4j.driver.internal;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.neo4j.driver.internal.retry.RetryDecision;
+import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 import org.neo4j.driver.internal.spi.PooledConnection;
@@ -45,6 +49,7 @@ public class NetworkSession implements Session, SessionResourcesHandler
 {
     private final ConnectionProvider connectionProvider;
     private final AccessMode mode;
+    private final RetryLogic<RetryDecision> retryLogic;
     protected final Logger logger;
 
     private String lastBookmark;
@@ -53,10 +58,12 @@ public class NetworkSession implements Session, SessionResourcesHandler
 
     private final AtomicBoolean isOpen = new AtomicBoolean( true );
 
-    public NetworkSession( ConnectionProvider connectionProvider, AccessMode mode, Logging logging )
+    public NetworkSession( ConnectionProvider connectionProvider, AccessMode mode, RetryLogic<RetryDecision> retryLogic,
+            Logging logging )
     {
         this.connectionProvider = connectionProvider;
         this.mode = mode;
+        this.retryLogic = retryLogic;
         this.logger = logging.getLog( "Session-" + hashCode() );
     }
 
@@ -242,9 +249,29 @@ public synchronized void onConnectionError( boolean recoverable )
 
     private synchronized <T> T transaction( AccessMode mode, Function<Transaction,T> work )
     {
-        try ( Transaction tx = beginTransaction( mode ) )
+        RetryDecision decision = null;
+        List<Throwable> errors = null;
+
+        while ( true )
         {
-            return work.apply( tx );
+            try ( Transaction tx = beginTransaction( mode ) )
+            {
+                return work.apply( tx );
+            }
+            catch ( Throwable newError )
+            {
+                decision = retryLogic.apply( newError, decision );
+
+                if ( decision.shouldRetry() )
+                {
+                    errors = recordError( newError, errors );
+                }
+                else
+                {
+                    addSuppressed( newError, errors );
+                    throw newError;
+                }
+            }
         }
     }
 
@@ -348,4 +375,28 @@ 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/SessionFactoryImpl.java

@@ -18,6 +18,8 @@
  */
 package org.neo4j.driver.internal;
 
+import org.neo4j.driver.internal.retry.RetryDecision;
+import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.spi.ConnectionProvider;
 import org.neo4j.driver.v1.AccessMode;
 import org.neo4j.driver.v1.Config;
@@ -27,14 +29,16 @@
 public class SessionFactoryImpl implements SessionFactory
 {
     protected final ConnectionProvider connectionProvider;
+    protected final RetryLogic<RetryDecision> retryLogic;
     protected final Logging logging;
     protected final boolean leakedSessionsLoggingEnabled;
 
-    SessionFactoryImpl( ConnectionProvider connectionProvider, Config config, Logging logging )
+    SessionFactoryImpl( ConnectionProvider connectionProvider, RetryLogic<RetryDecision> retryLogic, Config config )
     {
         this.connectionProvider = connectionProvider;
         this.leakedSessionsLoggingEnabled = config.logLeakedSessions();
-        this.logging = logging;
+        this.retryLogic = retryLogic;
+        this.logging = config.logging();
     }
 
     @Override
@@ -43,11 +47,11 @@ public Session newInstance( AccessMode mode, String bookmark )
         NetworkSession session;
         if ( leakedSessionsLoggingEnabled )
         {
-            session = new LeakLoggingNetworkSession( connectionProvider, mode, logging );
+            session = new LeakLoggingNetworkSession( connectionProvider, mode, retryLogic, logging );
         }
         else
         {
-            session = new NetworkSession( connectionProvider, mode, logging );
+            session = new NetworkSession( connectionProvider, mode, retryLogic, logging );
         }
         session.setLastBookmark( bookmark );
         return session;

driver/src/main/java/org/neo4j/driver/internal/retry/RetryDecision.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.retry;
+
+public interface RetryDecision
+{
+    boolean shouldRetry();
+}

driver/src/main/java/org/neo4j/driver/internal/retry/RetryLogic.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.retry;
+
+public interface RetryLogic<T extends RetryDecision>
+{
+    T apply( Throwable error, T previousDecision );
+}

driver/src/main/java/org/neo4j/driver/internal/retry/RetrySettings.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.retry;
+
+public final class RetrySettings
+{
+    public static final int DEFAULT_MAX_ATTEMPTS = 3;
+    public static final int DEFAULT_DELAY_MS = 2_000;
+    public static final RetrySettings DEFAULT = new RetrySettings( DEFAULT_MAX_ATTEMPTS, DEFAULT_DELAY_MS );
+
+    private final int maxAttempts;
+    private final long delayMs;
+
+    public RetrySettings( int maxAttempts, long delayMs )
+    {
+        this.maxAttempts = maxAttempts;
+        this.delayMs = delayMs;
+    }
+
+    public int maxAttempts()
+    {
+        return maxAttempts;
+    }
+
+    public long delayMs()
+    {
+        return delayMs;
+    }
+}