Allow tracking version context on transaction that executed by multiple

threads (transactions that goes over rest end point).
Re-check version context when cursor repositioned to its current page again.

Check version context state after eager result visiting during serialization
since both bolt and rest writers will try to load all properties of
nodes and relationships from database, if those are part of result,
to send those to client as well. And since those executed outside of query
we need to expand version check to verify that after streaming result
we still based on valid snapshot.
If dirty context will be noticed exception will be thrown since we can't
restart that query - its too late and client already get partial results for it.

community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/EagerResult.java

@@ -25,11 +25,14 @@
 import java.util.Map;
 
 import org.neo4j.graphdb.ExecutionPlanDescription;
+import org.neo4j.graphdb.NotFoundException;
 import org.neo4j.graphdb.Notification;
 import org.neo4j.graphdb.QueryExecutionType;
 import org.neo4j.graphdb.QueryStatistics;
 import org.neo4j.graphdb.ResourceIterator;
 import org.neo4j.graphdb.Result;
+import org.neo4j.io.pagecache.tracing.cursor.context.VersionContext;
+import org.neo4j.kernel.impl.query.QueryExecutionKernelException;
 
 import static java.lang.System.lineSeparator;
 
@@ -40,12 +43,14 @@ class EagerResult implements Result
 {
     private static final String ITEM_SEPARATOR = ", ";
     private final Result originalResult;
+    private final VersionContext versionContext;
     private final List<Map<String, Object>> queryResult = new ArrayList<>();
     private int cursor;
 
-    EagerResult( Result result )
+    EagerResult( Result result, VersionContext versionContext )
     {
         this.originalResult = result;
+        this.versionContext = versionContext;
     }
 
     public void consume()
@@ -145,9 +150,28 @@ public Iterable<Notification> getNotifications()
     public <VisitationException extends Exception> void accept( ResultVisitor<VisitationException> visitor )
             throws VisitationException
     {
-        for ( Map<String,Object> map : queryResult )
+        try
         {
-            visitor.visit( new MapRow( map ) );
+            for ( Map<String,Object> map : queryResult )
+            {
+                visitor.visit( new MapRow( map ) );
+            }
+            checkIfDirty();
+        }
+        catch ( NotFoundException e )
+        {
+            checkIfDirty();
+            throw e;
+        }
+    }
+
+    private void checkIfDirty()
+    {
+        if ( versionContext.isDirty() )
+        {
+            throw new QueryExecutionKernelException(
+                    new UnstableSnapshotException( "Unable to get clean data snapshot for query serialisation." ) )
+                    .asUserException();
         }
     }
 

community/cypher/cypher/src/main/java/org/neo4j/cypher/internal/javacompat/SnapshotExecutionEngine.java

@@ -80,7 +80,7 @@ protected Result executeWithRetries( String query, Map<String,Object> parameters
             attempt++;
             versionContext.initRead();
             Result result = executor.execute( query, parameters, context );
-            eagerResult = new EagerResult( result );
+            eagerResult = new EagerResult( result, versionContext );
             eagerResult.consume();
             dirtySnapshot = versionContext.isDirty();
             if ( dirtySnapshot && result.getQueryStatistics().containsUpdates() )

community/cypher/cypher/src/test/java/org/neo4j/cypher/internal/javacompat/EagerResultITTest.java

@@ -38,6 +38,8 @@
 import org.neo4j.graphdb.GraphDatabaseService;
 import org.neo4j.graphdb.Label;
 import org.neo4j.graphdb.Node;
+import org.neo4j.graphdb.NotFoundException;
+import org.neo4j.graphdb.QueryExecutionException;
 import org.neo4j.graphdb.QueryExecutionType;
 import org.neo4j.graphdb.ResourceIterator;
 import org.neo4j.graphdb.Result;
@@ -201,6 +203,30 @@ public void eagerResultVisit() throws Exception
         assertThat( values, Matchers.containsInAnyOrder( "d", "y" ) );
     }
 
+    @Test( expected = QueryExecutionException.class )
+    public void dirtyContextDuringResultVisitResultInUnstableSnapshotException() throws Exception
+    {
+        Result result = database.execute( "MATCH (n) RETURN n.c" );
+        List<String> values = new ArrayList<>();
+        result.accept( (Result.ResultVisitor<Exception>) row ->
+        {
+            testCursorContext.markAsDirty();
+            values.add( row.getString( "n.c" ) );
+            return false;
+        } );
+    }
+
+    @Test( expected = QueryExecutionException.class )
+    public void dirtyContextEntityNotFoundExceptionDuringResultVisitResultInUnstableSnapshotException() throws Exception
+    {
+        Result result = database.execute( "MATCH (n) RETURN n.c" );
+        result.accept( (Result.ResultVisitor<Exception>) row ->
+        {
+            testCursorContext.markAsDirty();
+            throw new NotFoundException( new RuntimeException() );
+        } );
+    }
+
     private String printToStream( Result result )
     {
         StringWriter stringWriter = new StringWriter();

community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/CursorPool.java

@@ -74,8 +74,7 @@ MuninnReadPageCursor takeReadCursor( long pageId, int pf_flags )
 
     private MuninnReadPageCursor createReadCursor( CursorSets cursorSets )
     {
-        MuninnReadPageCursor cursor = new MuninnReadPageCursor( cursorSets, victimPage, getPageCursorTracer(), versionContextSupplier
-                .getVersionContext() );
+        MuninnReadPageCursor cursor = new MuninnReadPageCursor( cursorSets, victimPage, getPageCursorTracer(), versionContextSupplier );
         cursor.initialiseFile( pagedFile );
         return cursor;
     }
@@ -98,8 +97,7 @@ MuninnWritePageCursor takeWriteCursor( long pageId, int pf_flags )
 
     private MuninnWritePageCursor createWriteCursor( CursorSets cursorSets )
     {
-        MuninnWritePageCursor cursor = new MuninnWritePageCursor( cursorSets, victimPage, getPageCursorTracer(), versionContextSupplier
-                .getVersionContext() );
+        MuninnWritePageCursor cursor = new MuninnWritePageCursor( cursorSets, victimPage, getPageCursorTracer(), versionContextSupplier );
         cursor.initialiseFile( pagedFile );
         return cursor;
     }

community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/MuninnPageCursor.java

@@ -30,6 +30,7 @@
 import org.neo4j.io.pagecache.tracing.PinEvent;
 import org.neo4j.io.pagecache.tracing.cursor.PageCursorTracer;
 import org.neo4j.io.pagecache.tracing.cursor.context.VersionContext;
+import org.neo4j.io.pagecache.tracing.cursor.context.VersionContextSupplier;
 import org.neo4j.unsafe.impl.internal.dragons.UnsafeUtil;
 
 import static org.neo4j.io.pagecache.PagedFile.PF_EAGER_FLUSH;
@@ -66,7 +67,7 @@ abstract class MuninnPageCursor extends PageCursor
     private long pointer;
     private int pageSize;
     private int filePageSize;
-    protected VersionContext versionContext;
+    protected final VersionContextSupplier versionContextSupplier;
     private int offset;
     private boolean outOfBounds;
     private boolean isLinkedCursor;
@@ -75,12 +76,12 @@ abstract class MuninnPageCursor extends PageCursor
     // offending code.
     private Object cursorException;
 
-    MuninnPageCursor( long victimPage, PageCursorTracer tracer, VersionContext versionContext )
+    MuninnPageCursor( long victimPage, PageCursorTracer tracer, VersionContextSupplier versionContextSupplier )
     {
         this.victimPage = victimPage;
         this.pointer = victimPage;
         this.tracer = tracer;
-        this.versionContext = versionContext;
+        this.versionContextSupplier = versionContextSupplier;
     }
 
     final void initialiseFile( MuninnPagedFile pagedFile )
@@ -119,12 +120,27 @@ public final boolean next( long pageId ) throws IOException
     {
         if ( currentPageId == pageId )
         {
+            verifyContext();
             return true;
         }
         nextPageId = pageId;
         return next();
     }
 
+    void verifyContext()
+    {
+        VersionContext versionContext = versionContextSupplier.getVersionContext();
+        if ( versionContext.lastClosedTransactionId() == Long.MAX_VALUE )
+        {
+            return;
+        }
+        if ( page.getLastModifiedTxId() > versionContext.lastClosedTransactionId() ||
+                pagedFile.getHighestEvictedTransactionId() > versionContext.lastClosedTransactionId() )
+        {
+            versionContext.markAsDirty();
+        }
+    }
+
     @Override
     public final void close()
     {

community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/MuninnReadPageCursor.java

@@ -23,7 +23,7 @@
 
 import org.neo4j.io.pagecache.PageSwapper;
 import org.neo4j.io.pagecache.tracing.cursor.PageCursorTracer;
-import org.neo4j.io.pagecache.tracing.cursor.context.VersionContext;
+import org.neo4j.io.pagecache.tracing.cursor.context.VersionContextSupplier;
 
 final class MuninnReadPageCursor extends MuninnPageCursor
 {
@@ -32,9 +32,9 @@ final class MuninnReadPageCursor extends MuninnPageCursor
     MuninnReadPageCursor nextCursor;
 
     MuninnReadPageCursor( CursorPool.CursorSets cursorSets, long victimPage, PageCursorTracer pageCursorTracer,
-            VersionContext versionContext )
+            VersionContextSupplier versionContextSupplier )
     {
-        super( victimPage, pageCursorTracer, versionContext );
+        super( victimPage, pageCursorTracer, versionContextSupplier );
         this.cursorSets = cursorSets;
     }
 
@@ -65,18 +65,18 @@ public boolean next() throws IOException
         return true;
     }
 
-    private void verifyContext()
-    {
-        if ( versionContext.lastClosedTransactionId() == Long.MAX_VALUE )
-        {
-            return;
-        }
+//    private void verifyContext()
+//    {
+//        if ( versionContext.lastClosedTransactionId() == Long.MAX_VALUE )
+//        {
+//            return;
+//        }
 //        if ( page.getLastModifiedTxId() > versionContext.lastClosedTransactionId() ||
 //             pagedFile.getHighestEvictedTransactionId() > versionContext.lastClosedTransactionId() )
 //        {
 //            versionContext.markAsDirty();
 //        }
-    }
+//    }
 
     @Override
     protected boolean tryLockPage( long pageRef )

community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/MuninnWritePageCursor.java

@@ -24,17 +24,17 @@
 import org.neo4j.io.pagecache.PageSwapper;
 import org.neo4j.io.pagecache.PagedFile;
 import org.neo4j.io.pagecache.tracing.cursor.PageCursorTracer;
-import org.neo4j.io.pagecache.tracing.cursor.context.VersionContext;
+import org.neo4j.io.pagecache.tracing.cursor.context.VersionContextSupplier;
 
 final class MuninnWritePageCursor extends MuninnPageCursor
 {
     private final CursorPool.CursorSets cursorSets;
     MuninnWritePageCursor nextCursor;
 
     MuninnWritePageCursor( CursorPool.CursorSets cursorSets, long victimPage, PageCursorTracer pageCursorTracer,
-            VersionContext versionContext )
+            VersionContextSupplier versionContextSupplier )
     {
-        super( victimPage, pageCursorTracer, versionContext );
+        super( victimPage, pageCursorTracer, versionContextSupplier );
         this.cursorSets = cursorSets;
     }
 
@@ -126,7 +126,7 @@ protected void pinCursorToPage( long pageRef, long filePageId, PageSwapper swapp
         // be closed and the page lock will be released.
         assertPagedFileStillMappedAndGetIdOfLastPage();
         pagedFile.incrementUsage( pageRef );
-        pagedFile.setLastModifiedTxId( versionContext.committingTransactionId() );
+        pagedFile.setLastModifiedTxId( versionContextSupplier.getVersionContext().committingTransactionId() );
     }
 
     @Override

community/io/src/main/java/org/neo4j/io/pagecache/tracing/cursor/context/EmptyVersionContext.java

@@ -65,9 +65,4 @@ public boolean isDirty()
     {
         return false;
     }
-
-    @Override
-    public void clear()
-    {
-    }
 }

community/io/src/main/java/org/neo4j/io/pagecache/tracing/cursor/context/VersionContext.java

@@ -64,9 +64,4 @@ public interface VersionContext
      */
     boolean isDirty();
 
-    /**
-     * Clear transaction ids that read/write context was initialised with
-     */
-    void clear();
-
 }

community/io/src/test/java/org/neo4j/io/pagecache/impl/muninn/MuninnPageCacheTest.java

@@ -244,7 +244,7 @@ public void trackPageModificationTransactionId() throws Exception
     }
 
     @Test
-    public void pareModificationTrackingNoticeWriteFromAnotherThread() throws Exception
+    public void pageModificationTrackingNoticeWriteFromAnotherThread() throws Exception
     {
         TestVersionContext cursorContext = new TestVersionContext( () -> 0 );
         VersionContextSupplier versionContextSupplier = new ConfiguredVersionContextSupplier( cursorContext );
@@ -312,6 +312,28 @@ public void pageModificationTracksHighestModifierTransactionId() throws IOExcept
         }
     }
 
+    @Test
+    public void markCursorContextDirtyWhenRepositionCursorOnItsCurrentPage() throws IOException
+    {
+        TestVersionContext cursorContext = new TestVersionContext( () -> 3 );
+        VersionContextSupplier versionContextSupplier = new ConfiguredVersionContextSupplier( cursorContext );
+        MuninnPageCache pageCache =
+                createPageCache( fs, 2, 8, PageCacheTracer.NULL, PageCursorTracerSupplier.NULL, versionContextSupplier );
+
+        PagedFile pagedFile = pageCache.map( file( "a" ), 8 );
+        cursorContext.initRead();
+        try ( PageCursor cursor = pagedFile.io( 0, PF_SHARED_WRITE_LOCK ) )
+        {
+            assertTrue( cursor.next( 0 ) );
+            assertFalse( cursorContext.isDirty() );
+
+            ((MuninnPageCursor) cursor).page.setLastModifiedTxId( 17 );
+
+            assertTrue( cursor.next( 0 ) );
+            assertTrue( cursorContext.isDirty() );
+        }
+    }
+
     @Test
     public void markCursorContextAsDirtyWhenReadingDataFromMoreRecentTransactions() throws IOException
     {
@@ -651,11 +673,5 @@ public boolean isDirty()
         {
             return dirty;
         }
-
-        @Override
-        public void clear()
-        {
-
-        }
     }
 }