Add page cache test for rare race bug

Add a test for the case where vectored flushing can race with eviction and page faulting, such that the flusher reads a page out of its translation table,
but before it can grab a read lock on that page, the page gets evicted and then faulted into a different swapper+filePageId combination and marked as dirty.
Previously, we would only check if a page was loaded before adding it to the vector of pages to be written, but now we check that the page is *bound* to the
expected swapper and file page id, and we do this check after we've grabbed the read lock on the page. This prevents the aforementioned race from happening.

This fix was already committed in #f64422b5487cfcd36f7acce0e1a970043d54e4ff - in this commit, we've modified the
concurrentFlushingMustNotPutInterleavedDataIntoFile test, such that it is capable of catching this bug.

community/io/src/main/java/org/neo4j/io/pagecache/impl/SingleFilePageSwapper.java

@@ -28,9 +28,9 @@
 import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
+import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.OverlappingFileLockException;
-import java.nio.channels.FileChannel;
 
 import org.neo4j.io.fs.FileSystemAbstraction;
 import org.neo4j.io.fs.FileUtils;

community/io/src/test/java/org/neo4j/io/pagecache/PageCacheTest.java

@@ -3405,14 +3405,14 @@ public void concurrentFlushingMustNotPutInterleavedDataIntoFile() throws Excepti
         final RecordFormat recordFormat = new StandardRecordFormat();
         final int filePageCount = 100;
         RandomPageCacheTestHarness harness = new RandomPageCacheTestHarness();
-        harness.setConcurrencyLevel( 6 );
+        harness.setConcurrencyLevel( 16 );
         harness.setUseAdversarialIO( false );
         harness.setCachePageCount( 100 );
         harness.setFilePageCount( filePageCount );
         harness.setCachePageSize( pageCachePageSize );
         harness.setFilePageSize( pageCachePageSize );
         harness.setInitialMappedFiles( 3 );
-        harness.setCommandCount( 10000 );
+        harness.setCommandCount( 50_000 );
         harness.disableCommands( Command.MapFile, Command.UnmapFile, Command.ReadRecord );
         harness.setVerification( new Phase()
         {
@@ -3425,12 +3425,23 @@ public void run(
                     try ( PagedFile pf = pageCache.map( file, pageCachePageSize );
                           PageCursor cursor = pf.io( 0, PF_SHARED_LOCK ) )
                     {
-                        for ( int pageId = 0; pageId < filePageCount; pageId++ )
+                        for ( int pageId = 0; pageId < filePageCount && cursor.next(); pageId++ )
                         {
-                            cursor.next();
-                            recordFormat.assertRecordsWrittenCorrectly( cursor );
+                            try
+                            {
+                                recordFormat.assertRecordsWrittenCorrectly( cursor );
+                            }
+                            catch ( Throwable th )
+                            {
+                                th.addSuppressed( new Exception( "pageId = " + pageId ) );
+                                throw th;
+                            }
                         }
                     }
+                    try ( StoreChannel channel = fs.open( file, "r" ) )
+                    {
+                        recordFormat.assertRecordsWrittenCorrectly( file, channel );
+                    }
                 }
             }
         } );

community/io/src/test/java/org/neo4j/io/pagecache/randomharness/RandomPageCacheTestHarness.java

@@ -398,10 +398,7 @@ private void runIteration( long timeout, TimeUnit unit ) throws Exception
                 }
                 future.get( deadlineMillis - now, TimeUnit.MILLISECONDS );
             }
-            if ( verification != null )
-            {
-                verification.run( cache, this.fs, plan.getFilesTouched() );
-            }
+            runVerificationPhase( cache );
         }
         finally
         {
@@ -419,6 +416,14 @@ private void runIteration( long timeout, TimeUnit unit ) throws Exception
         }
     }
 
+    private void runVerificationPhase( MuninnPageCache cache ) throws Exception
+    {
+        if ( verification != null )
+        {
+            verification.run( cache, this.fs, plan.getFilesTouched() );
+        }
+    }
+
     private File[] buildFileNames() throws IOException
     {
         String s = "abcdefghijklmnopqrstuvwxyz";

community/io/src/test/java/org/neo4j/io/pagecache/randomharness/RecordFormat.java

@@ -27,6 +27,7 @@
 import org.neo4j.io.pagecache.PageCursor;
 import org.neo4j.io.pagecache.StubPageCursor;
 
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertThat;
 
@@ -76,19 +77,34 @@ private void writeRecordToPage( PageCursor cursor, long pageId, int recordsPerPa
 
     public final void assertRecordsWrittenCorrectly( PageCursor cursor ) throws IOException
     {
-        int recordsPerPage = cursor.getCurrentPageSize() / getRecordSize();
+        int currentPageSize = cursor.getCurrentPageSize();
+        int recordSize = getRecordSize();
+        int recordsPerPage = currentPageSize / recordSize;
         for ( int pageRecordId = 0; pageRecordId < recordsPerPage; pageRecordId++ )
         {
-            int recordId = (int) (cursor.getCurrentPageId() * recordsPerPage + pageRecordId);
+            long currentPageId = cursor.getCurrentPageId();
+            int recordId = (int) (currentPageId * recordsPerPage + pageRecordId);
             Record expectedRecord = createRecord( cursor.getCurrentFile(), recordId );
             Record actualRecord;
-            int offset = cursor.getOffset();
-            do
-            {
-                cursor.setOffset( offset );
-                actualRecord = readRecord( cursor );
-            }
-            while ( cursor.shouldRetry() );
+            actualRecord = readRecord( cursor );
+            assertThat( actualRecord, isOneOf( expectedRecord, zeroRecord() ) );
+        }
+    }
+
+    public final void assertRecordsWrittenCorrectly( File file, StoreChannel channel ) throws IOException
+    {
+        int recordSize = getRecordSize();
+        long recordsInFile = channel.size() / recordSize;
+        ByteBuffer buffer = ByteBuffer.allocateDirect( recordSize );
+        StubPageCursor cursor = new StubPageCursor( 0, buffer );
+        for ( int i = 0; i < recordsInFile; i++ )
+        {
+            assertThat( "reading record id " + i, channel.read( buffer ), is( recordSize ) );
+            buffer.flip();
+            Record expectedRecord = createRecord( file, i );
+            cursor.setOffset( 0 );
+            Record actualRecord = readRecord( cursor );
+            buffer.clear();
             assertThat( actualRecord, isOneOf( expectedRecord, zeroRecord() ) );
         }
     }