Fixes issue with failure handling and recovery in ReplicatedTransacti…

…onSM

There was an issue with RTSM in combination with SessionTrackingState persistence,
 where a crash in between updating the session tracker and committing the tx in the
 logical log could lead to failure to actually commit the tx on recovery. This issue
 is now replicated in tests and fixed. The fix involves separating out the
 verification and update functionalities of GlobalSessionTrackerState, in order
 to invert the order of updating session state and committing in the logical log.
This commit also takes the opportunity to make sure that GlobalSessionTrackerState
 persistence is idempotent with respect to raft log entry replays.

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/session/GlobalSessionTrackerState.java

@@ -36,6 +36,9 @@ public interface GlobalSessionTrackerState<MEMBER>
     /**
      * Tracks the operation and returns true iff this operation should be allowed.
      */
-    boolean validateAndTrackOperationAtLogIndex( GlobalSession<MEMBER> globalSession, LocalOperationId localOperationId,
-                                                 long logIndex );
+    boolean validateOperation( GlobalSession<MEMBER> globalSession, LocalOperationId localOperationId );
+
+    void update( GlobalSession<MEMBER> globalSession, LocalOperationId localOperationId, long logIndex );
+
+    long logIndex();
 }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/session/InMemoryGlobalSessionTrackerState.java

@@ -56,15 +56,44 @@ public InMemoryGlobalSessionTrackerState( InMemoryGlobalSessionTrackerState<MEMB
     }
 
     @Override
-    public boolean validateAndTrackOperationAtLogIndex( GlobalSession<MEMBER> globalSession,
-                                                        LocalOperationId localOperationId, long logIndex )
+    public boolean validateOperation( GlobalSession<MEMBER> globalSession,
+                                      LocalOperationId localOperationId )
+    {
+        boolean result;
+
+        LocalSessionTracker existingSessionTracker = sessionTrackers.get( globalSession.owner() );
+        if ( isNewSession( globalSession, existingSessionTracker ) )
+        {
+            result = isFirstOperation( localOperationId );
+        }
+        else
+        {
+            result = existingSessionTracker.isValidOperation( localOperationId );
+        }
+
+        return result;
+    }
+
+    @Override
+    public void update( GlobalSession<MEMBER> globalSession, LocalOperationId localOperationId, long logIndex )
     {
-        this.logIndex = logIndex;
         LocalSessionTracker localSessionTracker = validateGlobalSessionAndGetLocalSessionTracker( globalSession );
-        return localSessionTracker.validateAndTrackOperation( localOperationId );
+        localSessionTracker.validateAndTrackOperation( localOperationId );
+        this.logIndex = logIndex;
     }
 
-    long logIndex()
+    private boolean isNewSession( GlobalSession<MEMBER> globalSession, LocalSessionTracker existingSessionTracker )
+    {
+        return existingSessionTracker == null || !existingSessionTracker.globalSessionId.equals( globalSession.sessionId() );
+    }
+
+    private boolean isFirstOperation( LocalOperationId id )
+    {
+        return id.sequenceNumber() == 0;
+    }
+
+    @Override
+    public long logIndex()
     {
         return logIndex;
     }
@@ -73,12 +102,8 @@ private LocalSessionTracker validateGlobalSessionAndGetLocalSessionTracker( Glob
     {
         LocalSessionTracker localSessionTracker = sessionTrackers.get( globalSession.owner() );
 
-        if ( localSessionTracker == null )
-        {
-            localSessionTracker = new LocalSessionTracker( globalSession.sessionId() );
-            sessionTrackers.put( globalSession.owner(), localSessionTracker );
-        }
-        else if ( !localSessionTracker.globalSessionId.equals( globalSession.sessionId() ) )
+        if ( localSessionTracker == null ||
+                !localSessionTracker.globalSessionId.equals( globalSession.sessionId() ) )
         {
             localSessionTracker = new LocalSessionTracker( globalSession.sessionId() );
             sessionTrackers.put( globalSession.owner(), localSessionTracker );

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/session/OnDiskGlobalSessionTrackerState.java

@@ -65,27 +65,35 @@ public OnDiskGlobalSessionTrackerState( FileSystemAbstraction fileSystemAbstract
     }
 
     @Override
-    public boolean validateAndTrackOperationAtLogIndex( GlobalSession<MEMBER> globalSession, LocalOperationId
-            localOperationId, long logIndex )
+    public boolean validateOperation( GlobalSession<MEMBER> globalSession, LocalOperationId
+            localOperationId )
+    {
+        return inMemoryGlobalSessionTrackerState.validateOperation( globalSession, localOperationId );
+    }
+
+    @Override
+    public void update( GlobalSession<MEMBER> globalSession, LocalOperationId localOperationId, long logIndex )
     {
         InMemoryGlobalSessionTrackerState<MEMBER> temp =
                 new InMemoryGlobalSessionTrackerState<>( inMemoryGlobalSessionTrackerState );
 
-        boolean stateUpdated = temp.validateAndTrackOperationAtLogIndex( globalSession, localOperationId, logIndex );
+        temp.update( globalSession, localOperationId, logIndex );
 
-        if ( stateUpdated )
+        try
+        {
+            statePersister.persistStoreData( temp );
+            inMemoryGlobalSessionTrackerState = temp;
+        }
+        catch ( IOException e )
         {
-            try
-            {
-                statePersister.persistStoreData( temp );
-                inMemoryGlobalSessionTrackerState = temp;
-            }
-            catch ( IOException e )
-            {
-                throw new RuntimeException( e );
-            }
+            throw new RuntimeException( e );
         }
-        return stateUpdated;
+    }
+
+    @Override
+    public long logIndex()
+    {
+        return inMemoryGlobalSessionTrackerState.logIndex();
     }
 
     @Override

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/ReplicatedTransactionFactory.java

@@ -43,8 +43,8 @@
 
 public class ReplicatedTransactionFactory
 {
-    public static ReplicatedTransaction createImmutableReplicatedTransaction(
-            TransactionRepresentation tx, GlobalSession globalSession, LocalOperationId localOperationId ) throws IOException
+    public static <T> ReplicatedTransaction<T> createImmutableReplicatedTransaction(
+            TransactionRepresentation tx, GlobalSession<T> globalSession, LocalOperationId localOperationId ) throws IOException
     {
         ByteBuf transactionBuffer = Unpooled.buffer();
 
@@ -58,7 +58,7 @@ public static ReplicatedTransaction createImmutableReplicatedTransaction(
         byte[] txBytes = Arrays.copyOf( transactionBuffer.array(), transactionBuffer.writerIndex() );
         transactionBuffer.release();
 
-        return new ReplicatedTransaction( txBytes, globalSession, localOperationId );
+        return new ReplicatedTransaction<>( txBytes, globalSession, localOperationId );
     }
 
     public static TransactionRepresentation extractTransactionRepresentation( ReplicatedTransaction replicatedTransaction, byte[] extraHeader ) throws IOException

enterprise/core-edge/src/main/java/org/neo4j/coreedge/raft/replication/tx/ReplicatedTransactionStateMachine.java

@@ -26,7 +26,6 @@
 import org.neo4j.coreedge.raft.replication.Replicator;
 import org.neo4j.coreedge.raft.replication.session.GlobalSession;
 import org.neo4j.coreedge.raft.replication.session.GlobalSessionTrackerState;
-import org.neo4j.coreedge.server.CoreMember;
 import org.neo4j.coreedge.server.core.locks.LockTokenManager;
 import org.neo4j.kernel.api.exceptions.TransactionFailureException;
 import org.neo4j.kernel.impl.api.TransactionCommitProcess;
@@ -39,9 +38,9 @@
 import static org.neo4j.coreedge.raft.replication.tx.LogIndexTxHeaderEncoding.encodeLogIndexAsTxHeader;
 import static org.neo4j.kernel.api.exceptions.Status.Transaction.LockSessionInvalid;
 
-public class ReplicatedTransactionStateMachine implements Replicator.ReplicatedContentListener
+public class ReplicatedTransactionStateMachine<MEMBER> implements Replicator.ReplicatedContentListener
 {
-    private final GlobalSessionTrackerState sessionTracker;
+    private final GlobalSessionTrackerState<MEMBER> sessionTracker;
     private final GlobalSession myGlobalSession;
     private final LockTokenManager lockTokenManager;
     private final TransactionCommitProcess commitProcess;
@@ -52,7 +51,7 @@ public ReplicatedTransactionStateMachine( TransactionCommitProcess commitProcess
                                               GlobalSession myGlobalSession,
                                               LockTokenManager lockTokenManager,
                                               CommittingTransactions transactionFutures,
-                                              GlobalSessionTrackerState globalSessionTrackerState )
+                                              GlobalSessionTrackerState<MEMBER> globalSessionTrackerState )
     {
         this.commitProcess = commitProcess;
         this.myGlobalSession = myGlobalSession;
@@ -66,66 +65,95 @@ public synchronized void onReplicated( ReplicatedContent content, long logIndex
     {
         if ( content instanceof ReplicatedTransaction )
         {
-            handleTransaction( (ReplicatedTransaction<CoreMember>) content, logIndex );
+            handleTransaction( (ReplicatedTransaction<MEMBER>) content, logIndex );
         }
     }
 
-    private void handleTransaction( ReplicatedTransaction<CoreMember> replicatedTx, long logIndex )
+    private void handleTransaction( ReplicatedTransaction<MEMBER> replicatedTx, long logIndex )
     {
-        if ( !sessionTracker.validateAndTrackOperationAtLogIndex( replicatedTx.globalSession(),
-                replicatedTx.localOperationId(), logIndex ) || logIndex <= lastCommittedIndex )
+        /*
+         * This check quickly verifies that the session is invalid. Since we update the session state *after* appending
+         * the tx to the log, we are certain here that on replay, if the session tracker says that the session is invalid,
+         * then the transaction either should never be committed or has already been appended in the log.
+         */
+        if ( !operationValid( replicatedTx ) )
         {
             return;
         }
 
-        TransactionRepresentation tx;
-        try
+        /*
+         * At this point, we need to check if the tx exists in the log. If it does, it is ok to skip it. However, we
+         * may still need to persist the session state (as we may crashed in between), which happens outside this
+         * if check.
+         */
+        if ( !txAlreadyCommitted( logIndex ) )
         {
-            byte[] extraHeader = encodeLogIndexAsTxHeader( logIndex );
-            tx = ReplicatedTransactionFactory.extractTransactionRepresentation(
-                    replicatedTx, extraHeader );
-        }
-        catch ( IOException e )
-        {
-            throw new IllegalStateException( "Failed to locally commit a transaction that has already been " +
-                    "committed to the RAFT log. This server cannot process later transactions and needs to be " +
-                    "restarted once the underlying cause has been addressed.", e );
-        }
+            TransactionRepresentation tx;
+            try
+            {
+                byte[] extraHeader = encodeLogIndexAsTxHeader( logIndex );
+                tx = ReplicatedTransactionFactory.extractTransactionRepresentation(
+                        replicatedTx, extraHeader );
+            }
+            catch ( IOException e )
+            {
+                throw new IllegalStateException( "Failed to locally commit a transaction that has already been " +
+                        "committed to the RAFT log. This server cannot process later transactions and needs to be " +
+                        "restarted once the underlying cause has been addressed.", e );
+            }
 
-        // A missing future means the transaction does not belong to this instance
-        Optional<CommittingTransaction> future = replicatedTx.globalSession().equals( myGlobalSession ) ?
-                Optional.ofNullable( transactionFutures.retrieve( replicatedTx.localOperationId() ) ) :
-                Optional.<CommittingTransaction>empty();
+            // A missing future means the transaction does not belong to this instance
+            Optional<CommittingTransaction> future = replicatedTx.globalSession().equals( myGlobalSession ) ?
+                    Optional.ofNullable( transactionFutures.retrieve( replicatedTx.localOperationId() ) ) :
+                    Optional.<CommittingTransaction>empty();
 
-        int currentTokenId = lockTokenManager.currentToken().id();
-        int txLockSessionId = tx.getLockSessionId();
+            int currentTokenId = lockTokenManager.currentToken().id();
+            int txLockSessionId = tx.getLockSessionId();
 
-        if ( currentTokenId != txLockSessionId && txLockSessionId != Locks.Client.NO_LOCK_SESSION_ID )
-        {
-            future.ifPresent( txFuture -> txFuture.notifyCommitFailed( new TransactionFailureException(
-                    LockSessionInvalid,
-                    "The lock session in the cluster has changed: " +
-                            "[current lock session id:%d, tx lock session id:%d]",
-                    currentTokenId, txLockSessionId ) ) );
-            return;
-        }
+            if ( currentTokenId != txLockSessionId && txLockSessionId != Locks.Client.NO_LOCK_SESSION_ID )
+            {
+                future.ifPresent( txFuture -> txFuture.notifyCommitFailed( new TransactionFailureException(
+                        LockSessionInvalid,
+                        "The lock session in the cluster has changed: " +
+                                "[current lock session id:%d, tx lock session id:%d]",
+                        currentTokenId, txLockSessionId ) ) );
+                return;
+            }
 
-        try
-        {
-            long txId = commitProcess.commit( new TransactionToApply( tx ), CommitEvent.NULL,
-                    TransactionApplicationMode.EXTERNAL );
+            try
+            {
+                long txId = commitProcess.commit( new TransactionToApply( tx ), CommitEvent.NULL,
+                        TransactionApplicationMode.EXTERNAL );
 
-            future.ifPresent( txFuture -> txFuture.notifySuccessfullyCommitted( txId ) );
+                future.ifPresent( txFuture -> txFuture.notifySuccessfullyCommitted( txId ) );
+            }
+            catch ( TransactionFailureException e )
+            {
+                future.ifPresent( txFuture -> txFuture.notifyCommitFailed( e ) );
+                throw new IllegalStateException( "Failed to locally commit a transaction that has already been " +
+                        "committed to the RAFT log. This server cannot process later transactions and needs to be " +
+                        "restarted once the underlying cause has been addressed.", e );
+            }
         }
-        catch ( TransactionFailureException e )
+        /*
+         * Finally, we need to check, in an idempotent fashion, if the session state needs to be persisted.
+         */
+        if ( sessionTracker.logIndex() < logIndex )
         {
-            future.ifPresent( txFuture -> txFuture.notifyCommitFailed( e ) );
-            throw new IllegalStateException( "Failed to locally commit a transaction that has already been " +
-                    "committed to the RAFT log. This server cannot process later transactions and needs to be " +
-                    "restarted once the underlying cause has been addressed.", e );
+            sessionTracker.update( replicatedTx.globalSession(), replicatedTx.localOperationId(), logIndex );
         }
     }
 
+    private boolean operationValid( ReplicatedTransaction<MEMBER> replicatedTx )
+    {
+        return sessionTracker.validateOperation( replicatedTx.globalSession(), replicatedTx.localOperationId() );
+    }
+
+    private boolean txAlreadyCommitted( long logIndex )
+    {
+        return logIndex <= lastCommittedIndex;
+    }
+
     public void setLastCommittedIndex( long lastCommittedIndex )
     {
         this.lastCommittedIndex = lastCommittedIndex;

enterprise/core-edge/src/main/java/org/neo4j/coreedge/server/core/EnterpriseCoreEditionModule.java

@@ -403,15 +403,15 @@ public static CommitProcessFactory createCommitProcessFactory( final Replicator
                                                                    final Dependencies dependencies,
                                                                    final LogService logging,
                                                                    Monitors monitors,
-                                                                   GlobalSessionTrackerState globalSessionTrackerState )
+                                                                   GlobalSessionTrackerState<CoreMember> globalSessionTrackerState )
     {
         return ( appender, applier, config ) -> {
             TransactionRepresentationCommitProcess localCommit =
                     new TransactionRepresentationCommitProcess( appender, applier );
             dependencies.satisfyDependencies( localCommit );
 
             CommittingTransactions committingTransactions = new CommittingTransactionsRegistry();
-            ReplicatedTransactionStateMachine replicatedTxStateMachine = new ReplicatedTransactionStateMachine(
+            ReplicatedTransactionStateMachine<CoreMember> replicatedTxStateMachine = new ReplicatedTransactionStateMachine<>(
                     localCommit, localSessionPool.getGlobalSession(), currentReplicatedLockState,
                     committingTransactions, globalSessionTrackerState );