Added authorization expired response handling

driver/src/main/java/org/neo4j/driver/exceptions/AuthorizationExpiredException.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.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.exceptions;
+
+/**
+ * The authorization info maintained on the server has expired. The client should reconnect.
+ * <p>
+ * Error code: Neo.ClientError.Security.AuthorizationExpired
+ */
+public class AuthorizationExpiredException extends SecurityException
+{
+    public AuthorizationExpiredException( String code, String message )
+    {
+        super( code, message );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/async/connection/AuthorizationStateListener.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.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.async.connection;
+
+import io.netty.channel.Channel;
+
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+
+/**
+ * Listener for authorization info state maintained on the server side.
+ */
+public interface AuthorizationStateListener
+{
+    /**
+     * Notifies the listener that the credentials stored on the server side have expired.
+     *
+     * @param e       the {@link AuthorizationExpiredException} exception.
+     * @param channel the channel that received the error.
+     */
+    void onExpired( AuthorizationExpiredException e, Channel channel );
+}

driver/src/main/java/org/neo4j/driver/internal/async/connection/ChannelAttributes.java

@@ -40,6 +40,7 @@ public final class ChannelAttributes
     private static final AttributeKey<Long> LAST_USED_TIMESTAMP = newInstance( "lastUsedTimestamp" );
     private static final AttributeKey<InboundMessageDispatcher> MESSAGE_DISPATCHER = newInstance( "messageDispatcher" );
     private static final AttributeKey<String> TERMINATION_REASON = newInstance( "terminationReason" );
+    private static final AttributeKey<AuthorizationStateListener> AUTHORIZATION_STATE_LISTENER = newInstance( "authorizationStateListener" );
 
     private ChannelAttributes()
     {
@@ -145,6 +146,16 @@ public static void setTerminationReason( Channel channel, String reason )
         setOnce( channel, TERMINATION_REASON, reason );
     }
 
+    public static AuthorizationStateListener authorizationStateListener( Channel channel )
+    {
+        return get( channel, AUTHORIZATION_STATE_LISTENER );
+    }
+
+    public static void setAuthorizationStateListener( Channel channel, AuthorizationStateListener authorizationStateListener )
+    {
+        set( channel, AUTHORIZATION_STATE_LISTENER, authorizationStateListener );
+    }
+
     private static <T> T get( Channel channel, AttributeKey<T> key )
     {
         return channel.attr( key ).get();

driver/src/main/java/org/neo4j/driver/internal/async/inbound/InboundMessageDispatcher.java

@@ -25,18 +25,19 @@
 import java.util.Map;
 import java.util.Queue;
 
-import org.neo4j.driver.exceptions.ServiceUnavailableException;
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.Logging;
+import org.neo4j.driver.Value;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.internal.handlers.ResetResponseHandler;
 import org.neo4j.driver.internal.logging.ChannelActivityLogger;
 import org.neo4j.driver.internal.messaging.ResponseMessageHandler;
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.ErrorUtil;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.Logging;
-import org.neo4j.driver.Value;
-import org.neo4j.driver.exceptions.ClientException;
 
 import static java.util.Objects.requireNonNull;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.authorizationStateListener;
 import static org.neo4j.driver.internal.messaging.request.ResetMessage.RESET;
 import static org.neo4j.driver.internal.util.ErrorUtil.addSuppressed;
 
@@ -114,9 +115,17 @@ public void handleFailureMessage( String code, String message )
             return;
         }
 
-        // write a RESET to "acknowledge" the failure
-        enqueue( new ResetResponseHandler( this ) );
-        channel.writeAndFlush( RESET, channel.voidPromise() );
+        Throwable currentError = this.currentError;
+        if ( currentError instanceof AuthorizationExpiredException )
+        {
+            authorizationStateListener( channel ).onExpired( (AuthorizationExpiredException) currentError, channel );
+        }
+        else
+        {
+            // write a RESET to "acknowledge" the failure
+            enqueue( new ResetResponseHandler( this ) );
+            channel.writeAndFlush( RESET, channel.voidPromise() );
+        }
 
         ResponseHandler handler = removeHandler();
         handler.onFailure( currentError );

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -46,6 +46,7 @@
 import org.neo4j.driver.internal.util.Futures;
 
 import static java.lang.String.format;
+import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setAuthorizationStateListener;
 import static org.neo4j.driver.internal.util.Futures.combineErrors;
 import static org.neo4j.driver.internal.util.Futures.completeWithNullIfNoError;
 
@@ -66,19 +67,22 @@ public class ConnectionPoolImpl implements ConnectionPool
     private final ConnectionFactory connectionFactory;
 
     public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings, MetricsListener metricsListener, Logging logging,
-            Clock clock, boolean ownsEventLoopGroup )
+                               Clock clock, boolean ownsEventLoopGroup )
     {
-        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ), settings, metricsListener, logging,
-                clock, ownsEventLoopGroup, new NetworkConnectionFactory( clock, metricsListener ) );
+        this( connector, bootstrap, new NettyChannelTracker( metricsListener, bootstrap.config().group().next(), logging ),
+              new NettyChannelHealthChecker( settings, clock, logging ), settings, metricsListener, logging,
+              clock, ownsEventLoopGroup, new NetworkConnectionFactory( clock, metricsListener ) );
     }
 
-    public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker, PoolSettings settings,
-            MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup, ConnectionFactory connectionFactory )
+    protected ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker,
+                                  NettyChannelHealthChecker nettyChannelHealthChecker, PoolSettings settings,
+                                  MetricsListener metricsListener, Logging logging, Clock clock, boolean ownsEventLoopGroup,
+                                  ConnectionFactory connectionFactory )
     {
         this.connector = connector;
         this.bootstrap = bootstrap;
         this.nettyChannelTracker = nettyChannelTracker;
-        this.channelHealthChecker = new NettyChannelHealthChecker( settings, clock, logging );
+        this.channelHealthChecker = nettyChannelHealthChecker;
         this.settings = settings;
         this.metricsListener = metricsListener;
         this.log = logging.getLog( ConnectionPool.class.getSimpleName() );
@@ -104,6 +108,7 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
             {
                 processAcquisitionError( pool, address, error );
                 assertNotClosed( address, channel, pool );
+                setAuthorizationStateListener( channel, channelHealthChecker );
                 Connection connection = connectionFactory.createConnection( channel, pool );
 
                 metricsListener.afterAcquiredOrCreated( pool.id(), acquireEvent );

driver/src/main/java/org/neo4j/driver/internal/async/pool/NettyChannelHealthChecker.java

@@ -23,27 +23,34 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
 
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.neo4j.driver.Logger;
+import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
+import org.neo4j.driver.internal.async.connection.AuthorizationStateListener;
 import org.neo4j.driver.internal.handlers.PingResponseHandler;
 import org.neo4j.driver.internal.messaging.request.ResetMessage;
 import org.neo4j.driver.internal.util.Clock;
-import org.neo4j.driver.Logger;
-import org.neo4j.driver.Logging;
 
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.creationTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.lastUsedTimestamp;
 import static org.neo4j.driver.internal.async.connection.ChannelAttributes.messageDispatcher;
 
-public class NettyChannelHealthChecker implements ChannelHealthChecker
+public class NettyChannelHealthChecker implements ChannelHealthChecker, AuthorizationStateListener
 {
     private final PoolSettings poolSettings;
     private final Clock clock;
     private final Logger log;
+    private final AtomicReference<Optional<Long>> minCreationTimestampMillisOpt;
 
     public NettyChannelHealthChecker( PoolSettings poolSettings, Clock clock, Logging logging )
     {
         this.poolSettings = poolSettings;
         this.clock = clock;
         this.log = logging.getLog( getClass().getSimpleName() );
+        this.minCreationTimestampMillisOpt = new AtomicReference<>( Optional.empty() );
     }
 
     @Override
@@ -60,11 +67,27 @@ public Future<Boolean> isHealthy( Channel channel )
         return ACTIVE.isHealthy( channel );
     }
 
+    @Override
+    public void onExpired( AuthorizationExpiredException e, Channel channel )
+    {
+        long ts = creationTimestamp( channel );
+        // Override current value ONLY if the new one is greater
+        minCreationTimestampMillisOpt.getAndUpdate( prev -> Optional.of( prev.filter( prevTs -> ts <= prevTs ).orElse( ts ) ) );
+    }
+
     private boolean isTooOld( Channel channel )
     {
-        if ( poolSettings.maxConnectionLifetimeEnabled() )
+        long creationTimestampMillis = creationTimestamp( channel );
+        Optional<Long> minCreationTimestampMillisOpt = this.minCreationTimestampMillisOpt.get();
+
+        if ( minCreationTimestampMillisOpt.isPresent() && creationTimestampMillis <= minCreationTimestampMillisOpt.get() )
+        {
+            log.trace( "The channel %s is marked for closure as its creation timestamp is older than or equal to the acceptable minimum timestamp: %s <= %s",
+                       channel, creationTimestampMillis, minCreationTimestampMillisOpt.get() );
+            return true;
+        }
+        else if ( poolSettings.maxConnectionLifetimeEnabled() )
         {
-            long creationTimestampMillis = creationTimestamp( channel );
             long currentTimestampMillis = clock.millis();
 
             long ageMillis = currentTimestampMillis - creationTimestampMillis;
@@ -74,7 +97,7 @@ private boolean isTooOld( Channel channel )
             if ( tooOld )
             {
                 log.trace( "Failed acquire channel %s from the pool because it is too old: %s > %s",
-                        channel, ageMillis, maxAgeMillis );
+                           channel, ageMillis, maxAgeMillis );
             }
 
             return tooOld;

driver/src/main/java/org/neo4j/driver/internal/messaging/v3/BoltProtocolV3.java

@@ -21,15 +21,13 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelPromise;
 
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import org.neo4j.driver.AuthToken;
 import org.neo4j.driver.Bookmark;
 import org.neo4j.driver.Query;
 import org.neo4j.driver.TransactionConfig;
-import org.neo4j.driver.Value;
 import org.neo4j.driver.internal.BookmarkHolder;
 import org.neo4j.driver.internal.DatabaseName;
 import org.neo4j.driver.internal.async.UnmanagedTransaction;
@@ -123,19 +121,10 @@ public CompletionStage<Void> beginTransaction( Connection connection, Bookmark b
             return Futures.failedFuture( error );
         }
 
+        CompletableFuture<Void> beginTxFuture = new CompletableFuture<>();
         BeginMessage beginMessage = new BeginMessage( bookmark, config, connection.databaseName(), connection.mode() );
-
-        if ( bookmark.isEmpty() )
-        {
-            connection.write( beginMessage, NoOpResponseHandler.INSTANCE );
-            return Futures.completedWithNull();
-        }
-        else
-        {
-            CompletableFuture<Void> beginTxFuture = new CompletableFuture<>();
-            connection.writeAndFlush( beginMessage, new BeginTxResponseHandler( beginTxFuture ) );
-            return beginTxFuture;
-        }
+        connection.writeAndFlush( beginMessage, new BeginTxResponseHandler( beginTxFuture ) );
+        return beginTxFuture;
     }
 
     @Override

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

@@ -39,6 +39,7 @@
 
 import org.neo4j.driver.Logger;
 import org.neo4j.driver.Logging;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.ServiceUnavailableException;
 import org.neo4j.driver.exceptions.SessionExpiredException;
@@ -155,7 +156,8 @@ protected boolean canRetryOn( Throwable error )
     @Experimental
     public static boolean isRetryable( Throwable error )
     {
-        return error instanceof SessionExpiredException || error instanceof ServiceUnavailableException || isTransientError( error );
+        return error instanceof SessionExpiredException || error instanceof ServiceUnavailableException || error instanceof AuthorizationExpiredException ||
+               isTransientError( error );
     }
 
     /**

driver/src/main/java/org/neo4j/driver/internal/util/ErrorUtil.java

@@ -25,6 +25,7 @@
 import java.util.stream.Stream;
 
 import org.neo4j.driver.exceptions.AuthenticationException;
+import org.neo4j.driver.exceptions.AuthorizationExpiredException;
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.DatabaseException;
 import org.neo4j.driver.exceptions.FatalDiscoveryException;
@@ -75,6 +76,10 @@ else if ( code.equalsIgnoreCase( "Neo.ClientError.Database.DatabaseNotFound" ) )
             {
                 return new FatalDiscoveryException( code, message );
             }
+            else if ( code.equalsIgnoreCase( "Neo.ClientError.Security.AuthorizationExpired" ) )
+            {
+                return new AuthorizationExpiredException( code, message );
+            }
             else
             {
                 return new ClientException( code, message );