Make the inconsistency injection of the PageCacheRule more devious

Previously, the `PageCacheRule` would by default randomly decide that some reads would be inconsistent.
Then, for those inconsistent reads, the page cursor would return zeros for all read operations.

Now, we still randomly choose in the `PageCursor.next()` call if a read will be inconsistent, but instead of just returning zeros, we are now more devious:
To start with, we just delegate all read operations and let them succeed, counting them in the process.
Then, in the `shouldRetry` method, we choose an operation at random to be the start of the inconsistent read.
We then let the reader retry, and when they get to the victim operation, we start returning random data from that point on.

This surfaces a number of bugs, which have also been fixed in this commit.

community/io/src/test/java/org/neo4j/adversaries/pagecache/AdversarialReadPageCursor.java

@@ -22,7 +22,11 @@
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
 
 import org.neo4j.adversaries.Adversary;
 import org.neo4j.io.pagecache.PageCursor;
@@ -46,7 +50,13 @@ class AdversarialReadPageCursor implements PageCursor
     private final PageCursor delegate;
     private final Adversary adversary;
 
+    private boolean currentReadIsPreparingInconsistent;
     private boolean currentReadIsInconsistent;
+    private int callCounter;
+
+    // This field for meant to be inspected with the debugger.
+    @SuppressWarnings( "MismatchedQueryAndUpdateOfCollection" )
+    private List<Object> inconsistentReadHistory;
 
     AdversarialReadPageCursor( PageCursor delegate, Adversary adversary )
     {
@@ -57,13 +67,51 @@ class AdversarialReadPageCursor implements PageCursor
     @Override
     public byte getByte()
     {
-        return currentReadIsInconsistent ? 0 : delegate.getByte();
+        return inconsistently( delegate.getByte() ).byteValue();
+    }
+
+    private <T extends Number> Number inconsistently( T value )
+    {
+        if ( currentReadIsPreparingInconsistent )
+        {
+            callCounter++;
+            return value;
+        }
+        if ( currentReadIsInconsistent && (--callCounter) <= 0 )
+        {
+            ThreadLocalRandom rng = ThreadLocalRandom.current();
+            long x = value.longValue();
+            if ( x != 0 & rng.nextBoolean() )
+            {
+                x = ~x;
+            }
+            else
+            {
+                x = rng.nextLong();
+            }
+            inconsistentReadHistory.add( new NumberValue( value.getClass(), x, delegate.getOffset() ) );
+            return x;
+        }
+        return value;
+    }
+
+    private void inconsistently( byte[] data )
+    {
+        if ( currentReadIsPreparingInconsistent )
+        {
+            callCounter++;
+        }
+        else if ( currentReadIsInconsistent )
+        {
+            ThreadLocalRandom.current().nextBytes( data );
+            inconsistentReadHistory.add( Arrays.copyOf( data, data.length ) );
+        }
     }
 
     @Override
     public byte getByte( int offset )
     {
-        return currentReadIsInconsistent ? 0 : delegate.getByte( offset );
+        return inconsistently( delegate.getByte( offset ) ).byteValue();
     }
 
     @Override
@@ -81,13 +129,13 @@ public void putByte( int offset, byte value )
     @Override
     public long getLong()
     {
-        return currentReadIsInconsistent ? 0 : delegate.getLong();
+        return inconsistently( delegate.getLong() ).longValue();
     }
 
     @Override
     public long getLong( int offset )
     {
-        return currentReadIsInconsistent ? 0 : delegate.getLong( offset );
+        return inconsistently( delegate.getLong( offset ) ).longValue();
     }
 
     @Override
@@ -105,13 +153,13 @@ public void putLong( int offset, long value )
     @Override
     public int getInt()
     {
-        return currentReadIsInconsistent ? 0 : delegate.getInt();
+        return inconsistently( delegate.getInt() ).intValue();
     }
 
     @Override
     public int getInt( int offset )
     {
-        return currentReadIsInconsistent ? 0 : delegate.getInt( offset );
+        return inconsistently( delegate.getInt( offset ) ).intValue();
     }
 
     @Override
@@ -129,22 +177,20 @@ public void putInt( int offset, int value )
     @Override
     public long getUnsignedInt()
     {
-        return currentReadIsInconsistent ? 0 : delegate.getUnsignedInt();
+        return inconsistently( delegate.getUnsignedInt() ).longValue();
     }
 
     @Override
     public long getUnsignedInt( int offset )
     {
-        return currentReadIsInconsistent ? 0 : delegate.getUnsignedInt( offset );
+        return inconsistently( delegate.getUnsignedInt( offset ) ).longValue();
     }
 
     @Override
     public void getBytes( byte[] data )
     {
-        if ( !currentReadIsInconsistent )
-        {
-            delegate.getBytes( data );
-        }
+        delegate.getBytes( data );
+        inconsistently( data );
     }
 
     @Override
@@ -156,13 +202,13 @@ public void putBytes( byte[] data )
     @Override
     public short getShort()
     {
-        return currentReadIsInconsistent ? 0 : delegate.getShort();
+        return inconsistently( delegate.getShort() ).shortValue();
     }
 
     @Override
     public short getShort( int offset )
     {
-        return currentReadIsInconsistent ? 0 : delegate.getShort( offset );
+        return inconsistently( delegate.getShort( offset ) ).shortValue();
     }
 
     @Override
@@ -217,16 +263,18 @@ public void rewind()
     @Override
     public boolean next() throws IOException
     {
-        currentReadIsInconsistent = adversary.injectFailureOrMischief( FileNotFoundException.class, IOException.class,
+        currentReadIsPreparingInconsistent = adversary.injectFailureOrMischief( FileNotFoundException.class, IOException.class,
                 SecurityException.class, IllegalStateException.class );
+        callCounter = 0;
         return delegate.next();
     }
 
     @Override
     public boolean next( long pageId ) throws IOException
     {
-        currentReadIsInconsistent = adversary.injectFailureOrMischief( FileNotFoundException.class, IOException.class,
+        currentReadIsPreparingInconsistent = adversary.injectFailureOrMischief( FileNotFoundException.class, IOException.class,
                 SecurityException.class, IllegalStateException.class );
+        callCounter = 0;
         return delegate.next( pageId );
     }
 
@@ -241,6 +289,16 @@ public boolean shouldRetry() throws IOException
     {
         adversary.injectFailure( FileNotFoundException.class, IOException.class, SecurityException.class,
                 IllegalStateException.class );
+        if ( currentReadIsPreparingInconsistent )
+        {
+            currentReadIsPreparingInconsistent = false;
+            currentReadIsInconsistent = true;
+            callCounter = ThreadLocalRandom.current().nextInt( callCounter + 1 );
+            inconsistentReadHistory = new ArrayList<>();
+            delegate.shouldRetry();
+            delegate.setOffset( 0 );
+            return true;
+        }
         if ( currentReadIsInconsistent )
         {
             currentReadIsInconsistent = false;
@@ -250,4 +308,40 @@ public boolean shouldRetry() throws IOException
         }
         return delegate.shouldRetry();
     }
+
+    private static class NumberValue
+    {
+        private final Class<? extends Number> type;
+        private final Long value;
+        private final int offset;
+        private final Exception trace;
+
+        public NumberValue( Class<? extends Number> type, Long value, int offset )
+        {
+            this.type = type;
+            this.value = value;
+            this.offset = offset;
+            trace = new Exception( toString() );
+            trace.fillInStackTrace();
+        }
+
+        @Override
+        public String toString()
+        {
+            String typeName = type.getCanonicalName();
+            switch ( typeName )
+            {
+            case "java.lang.Byte": return "(byte)" + value.byteValue() + " at offset " + offset;
+            case "java.lang.Short": return "(short)" + value.shortValue() + " at offset " + offset;
+            case "java.lang.Integer": return "(int)" + value.intValue() + " at offset " + offset;
+            case "java.lang.Long": return "(long)" + value + " at offset " + offset;
+            }
+            return "(" + typeName + ")" + value + " at offset " + offset;
+        }
+
+        public void printStackTrace()
+        {
+            trace.printStackTrace();
+        }
+    }
 }

community/kernel/src/main/java/org/neo4j/kernel/impl/store/AbstractDynamicStore.java

@@ -186,7 +186,7 @@ public static ByteBuffer concatData( Collection<DynamicRecord> records, byte[] t
             totalSize += (record.getData().length - offset);
         }
         byte[] bArray = new byte[totalSize];
-        assert header != null : "header should be non-null since records should not be empty";
+        assert header != null : "header should be non-null since records should not be empty: " + records;
         int sourceOffset = header.length;
         int offset = 0;
         for ( byte[] currentArray : byteList )

community/kernel/src/main/java/org/neo4j/kernel/impl/store/CommonAbstractStore.java

@@ -649,11 +649,14 @@ public long scanForHighId() throws IOException
             long nextPageId = storeFile.getLastPageId();
             int recordsPerPage = recordsPerPage();
             int recordSize = getRecordSize();
+            long highestId = getNumberOfReservedLowIds();
             while ( nextPageId >= 0 && cursor.next( nextPageId ) )
             {
                 nextPageId--;
+                boolean found;
                 do
                 {
+                    found = false;
                     int currentRecord = recordsPerPage;
                     while ( currentRecord-- > 0 )
                     {
@@ -662,11 +665,17 @@ public long scanForHighId() throws IOException
                         if ( isRecordInUse( cursor ) )
                         {
                             // We've found the highest id in use
-                            return recordId + 1 /*+1 since we return the high id*/;
+                            found = true;
+                            highestId = recordId + 1; /*+1 since we return the high id*/;
+                            break;
                         }
                     }
                 }
                 while ( cursor.shouldRetry() );
+                if ( found )
+                {
+                    return highestId;
+                }
             }
 
             return getNumberOfReservedLowIds();

community/kernel/src/main/java/org/neo4j/kernel/impl/store/LongerShortString.java

@@ -801,9 +801,11 @@ public static String decode( PropertyBlock block )
 
     private static LongerShortString getEncodingTable( int encodingHeader )
     {
-        final LongerShortString encoding = ENCODINGS_BY_ENCODING[encodingHeader];
-        if (encoding==null) throw new IllegalArgumentException( "Invalid encoding '" + encoding + "'" );
-        return encoding;
+        if ( encodingHeader < 0 | ENCODINGS_BY_ENCODING.length <= encodingHeader )
+        {
+            return null;
+        }
+        return ENCODINGS_BY_ENCODING[encodingHeader];
     }
 
     private static Bits newBits( LongerShortString encoding, int length )
@@ -832,7 +834,10 @@ public static boolean writeLatin1Characters( String string, Bits bits )
         for ( int i = 0; i < length; i++ )
         {
             char c = string.charAt( i );
-            if ( c < 0 || c >= 256 ) return false;
+            if ( c >= 256 )
+            {
+                return false;
+            }
             bits.put( c, 8 ); // Just the lower byte
         }
         return true;
@@ -921,8 +926,18 @@ public static int calculateNumberOfBlocksUsed( long firstBlock )
         int encoding = bits.getByte( 5 );
         int length = bits.getByte( 6 );
         */
-        if (encoding==ENCODING_UTF8 || encoding == ENCODING_LATIN1) return calculateNumberOfBlocksUsedForStep8(length);
-        return calculateNumberOfBlocksUsed( getEncodingTable(encoding), length );
+        if ( encoding == ENCODING_UTF8 || encoding == ENCODING_LATIN1 )
+        {
+            return calculateNumberOfBlocksUsedForStep8(length);
+        }
+
+        LongerShortString encodingTable = getEncodingTable( encoding );
+        if ( encodingTable == null )
+        {
+            // We probably did an inconsistent read of the first block
+            return PropertyType.BLOCKS_USED_FOR_BAD_TYPE_OR_ENCODING;
+        }
+        return calculateNumberOfBlocksUsed( encodingTable, length );
     }
 
     public static int calculateNumberOfBlocksUsedForStep8( int length )

community/kernel/src/main/java/org/neo4j/kernel/impl/store/PropertyStore.java

@@ -340,6 +340,7 @@ record = getRecord( id, cursor );
             {
                 throw new InvalidRecordException( "PropertyRecord[" + id + "] not in use" );
             }
+            record.verifyRecordIsWellFormed();
 
             return record;
         }
@@ -400,7 +401,7 @@ private PropertyRecord getRecordFromBuffer( long id, PageCursor cursor )
 
         while ( cursor.getOffset() - offsetAtBeginning < RECORD_SIZE )
         {
-            PropertyBlock newBlock = getPropertyBlock( cursor );
+            PropertyBlock newBlock = getPropertyBlock( cursor, record );
             if ( newBlock != null )
             {
                 record.addPropertyBlock( newBlock );
@@ -427,7 +428,7 @@ private PropertyRecord getRecord( long id, PageCursor cursor )
      * result is returned, that has inUse() return false. Also, the argument is not
      * touched.
      */
-    private PropertyBlock getPropertyBlock( PageCursor cursor )
+    private PropertyBlock getPropertyBlock( PageCursor cursor, PropertyRecord record )
     {
         long header = cursor.getLong();
         PropertyType type = PropertyType.getPropertyType( header, true );
@@ -438,6 +439,19 @@ private PropertyBlock getPropertyBlock( PageCursor cursor )
         PropertyBlock toReturn = new PropertyBlock();
         // toReturn.setInUse( true );
         int numBlocks = type.calculateNumberOfBlocksUsed( header );
+        if ( numBlocks == PropertyType.BLOCKS_USED_FOR_BAD_TYPE_OR_ENCODING )
+        {
+            record.setMalformedMessage( "Invalid type or encoding of property block" );
+            return null;
+        }
+        if ( numBlocks > PropertyType.getPayloadSizeLongs() )
+        {
+            // We most likely got an inconsistent read, so return null to signal failure for now
+            record.setMalformedMessage(
+                    "I was given an array of size " + numBlocks +", but I wanted it to be " +
+                    PropertyType.getPayloadSizeLongs() );
+            return null;
+        }
         long[] blockData = new long[numBlocks];
         blockData[0] = header; // we already have that
         for ( int i = 1; i < numBlocks; i++ )

community/kernel/src/main/java/org/neo4j/kernel/impl/store/PropertyType.java

@@ -415,4 +415,5 @@ public byte[] readDynamicRecordHeader( byte[] recordBytes )
     }
 
     public static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
+    public static final int BLOCKS_USED_FOR_BAD_TYPE_OR_ENCODING = -1;
 }

community/kernel/src/main/java/org/neo4j/kernel/impl/store/ShortArray.java

@@ -728,7 +728,7 @@ public static Object decode( PropertyBlock block )
             requiredBits = 64;
         }
         ShortArray type = typeOf( (byte)typeId );
-        return type.createArray(arrayLength, bits, requiredBits);
+        return type == null? null : type.createArray(arrayLength, bits, requiredBits);
     }
 
 
@@ -767,7 +767,12 @@ public int getRequiredBits( long value )
 
     public static ShortArray typeOf( byte typeId )
     {
-        return ShortArray.values()[typeId-1];
+        ShortArray[] values = ShortArray.values();
+        if ( typeId <= 0 | typeId > values.length )
+        {
+            return null;
+        }
+        return values[typeId - 1];
     }
 
     public static ShortArray typeOf( Object array )