Added ability to enforce max connection lifetime

Driver keeps idle socket connections in a connection pool. These
connections can get invalidated while resting idle in the pool.
They can be killed by network equipment like load balancer, proxy or
firewall. It is also safer to refresh connections once in a while so
that database can renew corresponding resources.

This commit adds `maxConnectionLifetime` setting and makes driver
close too old connections. Checking and closing happens in an
application thread that tries to acquire a connection. Feature can
be used on it's own or in combination with
`connectionLivenessCheckTimeout` to guarantee validity of acquired
connections.

Author: lutovich

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

@@ -177,7 +177,7 @@ protected ConnectionPool createConnectionPool( AuthToken authToken, SecurityPlan
 
         ConnectionSettings connectionSettings = new ConnectionSettings( authToken, config.connectionTimeoutMillis() );
         PoolSettings poolSettings = new PoolSettings( config.maxIdleConnectionPoolSize(),
-                config.idleTimeBeforeConnectionTest() );
+                config.idleTimeBeforeConnectionTest(), config.maxConnectionLifetime() );
         Connector connector = createConnector( connectionSettings, securityPlan, config.logging() );
 
         return new SocketConnectionPool( poolSettings, connector, createClock(), config.logging() );

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

@@ -221,6 +221,12 @@ public BoltServerAddress boltServerAddress()
         return delegate.boltServerAddress();
     }
 
+    @Override
+    public long creationTimestamp()
+    {
+        return delegate.creationTimestamp();
+    }
+
     @Override
     public long lastUsedTimestamp()
     {

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

@@ -107,7 +107,7 @@ public PooledConnection acquire( Supplier<PooledConnection> supplier )
         return connection;
     }
 
-    void disposeBroken( PooledConnection connection )
+    void dispose( PooledConnection connection )
     {
         acquiredConnections.remove( connection );
         disposeSafely( connection );

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

@@ -21,17 +21,21 @@
 public class PoolSettings
 {
     public static final int NO_IDLE_CONNECTION_TEST = -1;
+    public static final int INFINITE_CONNECTION_LIFETIME = -1;
 
     public static final int DEFAULT_MAX_IDLE_CONNECTION_POOL_SIZE = 10;
     public static final int DEFAULT_IDLE_TIME_BEFORE_CONNECTION_TEST = NO_IDLE_CONNECTION_TEST;
+    public static final int DEFAULT_MAX_CONNECTION_LIFETIME = INFINITE_CONNECTION_LIFETIME;
 
     private final int maxIdleConnectionPoolSize;
     private final long idleTimeBeforeConnectionTest;
+    private final long maxConnectionLifetime;
 
-    public PoolSettings( int maxIdleConnectionPoolSize, long idleTimeBeforeConnectionTest )
+    public PoolSettings( int maxIdleConnectionPoolSize, long idleTimeBeforeConnectionTest, long maxConnectionLifetime )
     {
         this.maxIdleConnectionPoolSize = maxIdleConnectionPoolSize;
         this.idleTimeBeforeConnectionTest = idleTimeBeforeConnectionTest;
+        this.maxConnectionLifetime = maxConnectionLifetime;
     }
 
     public int maxIdleConnectionPoolSize()
@@ -53,4 +57,19 @@ public boolean idleTimeBeforeConnectionTestConfigured()
     {
         return idleTimeBeforeConnectionTest >= 0;
     }
+
+    public long maxConnectionLifetime()
+    {
+        if ( !maxConnectionLifetimeConfigured() )
+        {
+            throw new IllegalStateException(
+                    "Max connection lifetime is not configured: " + maxConnectionLifetime );
+        }
+        return maxConnectionLifetime;
+    }
+
+    public boolean maxConnectionLifetimeConfigured()
+    {
+        return maxConnectionLifetime > 0;
+    }
 }

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

@@ -47,7 +47,7 @@ public void accept( PooledConnection pooledConnection )
         }
         else
         {
-            connections.disposeBroken( pooledConnection );
+            connections.dispose( pooledConnection );
         }
     }
 }

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

@@ -60,13 +60,16 @@ public class PooledSocketConnection implements PooledConnection
     private boolean unrecoverableErrorsOccurred = false;
     private SessionResourcesHandler resourcesHandler;
     private final Clock clock;
+
+    private final long creationTimestamp;
     private long lastUsedTimestamp;
 
     public PooledSocketConnection( Connection delegate, Consumer<PooledConnection> release, Clock clock )
     {
         this.delegate = delegate;
         this.release = release;
         this.clock = clock;
+        this.creationTimestamp = clock.millis();
         updateLastUsedTimestamp();
     }
 
@@ -280,6 +283,12 @@ public void setResourcesHandler( SessionResourcesHandler resourcesHandler )
         this.resourcesHandler = resourcesHandler;
     }
 
+    @Override
+    public long creationTimestamp()
+    {
+        return creationTimestamp;
+    }
+
     @Override
     public long lastUsedTimestamp()
     {
@@ -315,6 +324,6 @@ private boolean isClientOrTransientError( RuntimeException e )
 
     private void updateLastUsedTimestamp()
     {
-        this.lastUsedTimestamp = clock.millis();
+        lastUsedTimestamp = clock.millis();
     }
 }

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

@@ -149,7 +149,7 @@ private PooledConnection acquireConnection( BoltServerAddress address,
             // dispose previous connection that can't be acquired
             if ( connection != null )
             {
-                connectionQueue.disposeBroken( connection );
+                connectionQueue.dispose( connection );
             }
 
             connection = connectionQueue.acquire( connectionSupplier );
@@ -162,18 +162,27 @@ private PooledConnection acquireConnection( BoltServerAddress address,
 
     private boolean canBeAcquired( PooledConnection connection, boolean connectionCreated )
     {
-        if ( poolSettings.idleTimeBeforeConnectionTestConfigured() )
+        if ( connectionCreated )
+        {
+            return true;
+        }
+
+        if ( poolSettings.maxConnectionLifetimeConfigured() )
         {
-            if ( connectionCreated )
+            if ( isTooOld( connection ) )
             {
-                return true;
+                return false;
             }
+        }
 
+        if ( poolSettings.idleTimeBeforeConnectionTestConfigured() )
+        {
             if ( hasBeenIdleForTooLong( connection ) )
             {
                 return connectionValidator.isConnected( connection );
             }
         }
+
         return true;
     }
 
@@ -183,6 +192,12 @@ private boolean hasBeenIdleForTooLong( PooledConnection connection )
         return idleTime > poolSettings.idleTimeBeforeConnectionTest();
     }
 
+    private boolean isTooOld( PooledConnection connection )
+    {
+        long lifetime = clock.millis() - connection.creationTimestamp();
+        return lifetime > poolSettings.maxConnectionLifetime();
+    }
+
     private void assertNotClosed( BoltServerAddress address, BlockingPooledConnectionQueue connections )
     {
         if ( closed.get() )

driver/src/main/java/org/neo4j/driver/internal/spi/PooledConnection.java

@@ -40,6 +40,13 @@ public interface PooledConnection extends Connection
      */
     boolean hasUnrecoverableErrors();
 
+    /**
+     * Timestamp of when this connection was created. This timestamp should never change.
+     *
+     * @return timestamp as returned by {@link Clock#millis()}.
+     */
+    long creationTimestamp();
+
     /**
      * Timestamp of when this connection was used. This timestamp is updated when connection is returned to the pool.
      *

driver/src/main/java/org/neo4j/driver/v1/Config.java

@@ -58,11 +58,8 @@ public class Config
 
     private final int maxIdleConnectionPoolSize;
 
-    /**
-     * Connections that have been idle in the pool longer than this threshold will
-     * be tested for validity before being returned to the user.
-     */
     private final long idleTimeBeforeConnectionTest;
+    private final long maxConnectionLifetime;
 
     /** Indicator for encrypted traffic */
     private final boolean encrypted;
@@ -83,6 +80,7 @@ private Config( ConfigBuilder builder)
         this.logLeakedSessions = builder.logLeakedSessions;
 
         this.idleTimeBeforeConnectionTest = builder.idleTimeBeforeConnectionTest;
+        this.maxConnectionLifetime = builder.maxConnectionLifetime;
         this.maxIdleConnectionPoolSize = builder.maxIdleConnectionPoolSize;
 
         this.encrypted = builder.encrypted;
@@ -143,6 +141,16 @@ public long idleTimeBeforeConnectionTest()
         return idleTimeBeforeConnectionTest;
     }
 
+    /**
+     * Pooled connections older than this threshold will be closed and removed from the pool.
+     *
+     * @return maximum lifetime in milliseconds
+     */
+    public long maxConnectionLifetime()
+    {
+        return maxConnectionLifetime;
+    }
+
     /**
      * @return the configured connection timeout value in milliseconds.
      */
@@ -223,6 +231,7 @@ public static class ConfigBuilder
         private boolean logLeakedSessions;
         private int maxIdleConnectionPoolSize = PoolSettings.DEFAULT_MAX_IDLE_CONNECTION_POOL_SIZE;
         private long idleTimeBeforeConnectionTest = PoolSettings.DEFAULT_IDLE_TIME_BEFORE_CONNECTION_TEST;
+        private long maxConnectionLifetime = PoolSettings.DEFAULT_MAX_CONNECTION_LIFETIME;
         private boolean encrypted = true;
         private TrustStrategy trustStrategy = trustAllCertificates();
         private LoadBalancingStrategy loadBalancingStrategy = LoadBalancingStrategy.LEAST_CONNECTED;
@@ -348,7 +357,7 @@ public ConfigBuilder withSessionLivenessCheckTimeout( long timeout )
          * Value {@code 0} means connections will always be tested for
          * validity and negative values mean connections will never be tested.
          *
-         * @param value the minimum idle time in milliseconds
+         * @param value the minimum idle time
          * @param unit the unit in which the duration is given
          * @return this builder
          */
@@ -358,6 +367,32 @@ public ConfigBuilder withConnectionLivenessCheckTimeout( long value, TimeUnit un
             return this;
         }
 
+        /**
+         * Pooled connections older than this threshold will be closed and removed from the pool. Such discarding
+         * happens during connection acquisition so that new session is never backed by an old connection.
+         * <p>
+         * Setting this option to a low value will cause a high connection churn and might result in a performance hit.
+         * <p>
+         * It is recommended to set maximum lifetime to a slightly smaller value than the one configured in network
+         * equipment (load balancer, proxy, firewall, etc. can also limit maximum connection lifetime).
+         * <p>
+         * Setting can also be used in combination with {@link #withConnectionLivenessCheckTimeout(long, TimeUnit)}. In
+         * this case, it is recommended to set liveness check to a value smaller than network equipment has and maximum
+         * lifetime to a reasonably large value to "renew" connections once in a while.
+         * <p>
+         * No maximum lifetime limit is imposed by default. Zero and negative values result in lifetime not being
+         * checked.
+         *
+         * @param value the maximum connection lifetime
+         * @param unit the unit in which the duration is given
+         * @return this builder
+         */
+        public ConfigBuilder withMaxConnectionLifetime( long value, TimeUnit unit )
+        {
+            this.maxConnectionLifetime = unit.toMillis( value );
+            return this;
+        }
+
         /**
          * Configure the {@link EncryptionLevel} to use, use this to control wether the driver uses TLS encryption or not.
          * @param level the TLS level to use

driver/src/test/java/org/neo4j/driver/internal/cluster/RoutingPooledConnectionErrorHandlingTest.java

@@ -59,6 +59,7 @@
 import static org.mockito.Mockito.when;
 import static org.neo4j.driver.internal.logging.DevNullLogging.DEV_NULL_LOGGING;
 import static org.neo4j.driver.internal.net.pooling.PoolSettings.DEFAULT_MAX_IDLE_CONNECTION_POOL_SIZE;
+import static org.neo4j.driver.internal.net.pooling.PoolSettings.INFINITE_CONNECTION_LIFETIME;
 import static org.neo4j.driver.internal.net.pooling.PoolSettings.NO_IDLE_CONNECTION_TEST;
 import static org.neo4j.driver.internal.spi.Collector.NO_OP;
 import static org.neo4j.driver.internal.util.Matchers.containsReader;
@@ -351,7 +352,8 @@ private static RoutingTable newRoutingTable( ClusterComposition clusterCompositi
     private static ConnectionPool newConnectionPool( Connector connector, BoltServerAddress... addresses )
     {
         int maxIdleConnections = DEFAULT_MAX_IDLE_CONNECTION_POOL_SIZE;
-        PoolSettings settings = new PoolSettings( maxIdleConnections, NO_IDLE_CONNECTION_TEST );
+        PoolSettings settings = new PoolSettings( maxIdleConnections, NO_IDLE_CONNECTION_TEST,
+                INFINITE_CONNECTION_LIFETIME );
         SocketConnectionPool pool = new SocketConnectionPool( settings, connector, Clock.SYSTEM, DEV_NULL_LOGGING );
 
         // force pool to create and memorize some connections

driver/src/test/java/org/neo4j/driver/internal/cluster/RoutingPooledConnectionTest.java

@@ -0,0 +1,59 @@
+/*
+ * 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.cluster;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import org.neo4j.driver.internal.spi.PooledConnection;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+@RunWith( MockitoJUnitRunner.class )
+public class RoutingPooledConnectionTest
+{
+    @Mock
+    private PooledConnection pooledConnection;
+    @InjectMocks
+    private RoutingPooledConnection routingPooledConnection;
+
+    @Test
+    public void shouldExposeCreationTimestamp()
+    {
+        when( pooledConnection.creationTimestamp() ).thenReturn( 42L );
+
+        long timestamp = routingPooledConnection.creationTimestamp();
+
+        assertEquals( 42L, timestamp );
+    }
+
+    @Test
+    public void shouldExposeLastUsedTimestamp()
+    {
+        when( pooledConnection.lastUsedTimestamp() ).thenReturn( 42L );
+
+        long timestamp = routingPooledConnection.lastUsedTimestamp();
+
+        assertEquals( 42L, timestamp );
+    }
+}

driver/src/test/java/org/neo4j/driver/internal/net/pooling/BlockingPooledConnectionQueueTest.java

@@ -287,15 +287,33 @@ public void shouldReportActiveConnections()
 
     @Test
     @SuppressWarnings( "unchecked" )
-    public void shouldDisposeBrokenConnections()
+    public void shouldDisposeConnections()
     {
         BlockingPooledConnectionQueue queue = newConnectionQueue( 5 );
 
         queue.offer( mock( PooledConnection.class ) );
         PooledConnection connection = queue.acquire( mock( Supplier.class ) );
         assertEquals( 1, queue.activeConnections() );
 
-        queue.disposeBroken( connection );
+        queue.dispose( connection );
+        assertEquals( 0, queue.activeConnections() );
+        verify( connection ).dispose();
+    }
+
+    @Test
+    @SuppressWarnings( "unchecked" )
+    public void shouldDisposeConnectionsThatThrowOnDisposal()
+    {
+        BlockingPooledConnectionQueue queue = newConnectionQueue( 5 );
+
+        PooledConnection pooledConnection = mock( PooledConnection.class );
+        doThrow( new RuntimeException() ).when( pooledConnection ).dispose();
+
+        queue.offer( pooledConnection );
+        PooledConnection connection = queue.acquire( mock( Supplier.class ) );
+        assertEquals( 1, queue.activeConnections() );
+
+        queue.dispose( connection );
         assertEquals( 0, queue.activeConnections() );
         verify( connection ).dispose();
     }