Merge pull request #2254 from adamretter/feature/journal-checksums

Journal Checksums

Author: dizzzz

.classpath

@@ -23,6 +23,7 @@
 	<classpathentry kind="src" path="extensions/webdav/src"/>
 	<classpathentry kind="src" path="extensions/commands/src"/>
 	<classpathentry kind="lib" path="lib/core/j8fu-1.21.jar"/>
+        <classpathentry kind="lib" path="lib/core/lz4-java-1.5.0.jar"/>
 	<classpathentry kind="lib" path="lib/core/icu4j-59_1.jar"/>
 	<classpathentry kind="lib" path="lib/core/icu4j-localespi-59_1.jar"/>
 	<classpathentry kind="lib" path="lib/core/pkg-java-fork.jar"/>

lib/core/lz4-java-1.5.0.jar

@@ -12,6 +12,7 @@ file.reference.commons-lang3-3.7.jar=lib/optional/commons-lang3-3.7.jar
 file.reference.commons-net-3.6.jar=lib/optional/commons-net-3.6.jar
 file.reference.httpcore-4.4.8.jar=lib/optional/httpcore-4.4.8.jar
 file.reference.j8fu-1.21.jar=lib/core/j8fu-1.21.jar
+file.reference.lz4-java-1.5.0.jar=lib/core/lz4-java-1.5.0.jar
 file.reference.icu4j-59_1.jar=lib/core/icu4j-59_1.jar
 file.reference.icu4j-localespi-59_1.jar=lib/core/icu4j-localespi-59_1.jar
 file.reference.caffeine-2.6.2.jar=lib/core/caffeine-2.6.2.jar
@@ -369,6 +370,7 @@ javac.classpath=\
     ${file.reference.tagsoup-1.2.1.jar}:\
     ${file.reference.exquery-annotations-common-1.0-SNAPSHOT.jar}:\
     ${file.reference.j8fu-1.21.jar}:\
+    ${file.reference.lz4-java-1.5.0.jar}:\
     ${file.reference.icu4j-59_1.jar}:\
     ${file.reference.icu4j-localespi-59_1.jar}:\
     ${file.reference.caffeine-2.6.2.jar}:\

nbproject/project.properties

@@ -65,6 +65,7 @@ lib/core/xmlrpc-server-%latest%.jar                always
 lib/core/clj-ds-%latest%.jar                       always
 lib/core/cglib-nodep-%latest%.jar                  always
 lib/core/j8fu-%latest%.jar			   always
+lib/core/lz4-java-%latest%.jar			   always
 lib/core/icu4j-%latest%.jar                        always
 lib/core/icu4j-localespi-%latest%.jar              always
 lib/core/caffeine-%latest%.jar			   always

src/org/exist/start/start.config

@@ -87,4 +87,12 @@ public void undo() throws LogException {
     public String dump() {
         return super.dump() + " - writing overflow page " + pageNum + "; next: " + nextPage;
     }
+
+    public long getPageNum() {
+        return pageNum;
+    }
+
+    public Value getValue() {
+        return value;
+    }
 }

src/org/exist/storage/dom/WriteOverflowPageLoggable.java

@@ -31,6 +31,8 @@
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import net.jpountz.xxhash.XXHash64;
+import net.jpountz.xxhash.XXHashFactory;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.exist.EXistException;
@@ -49,29 +51,50 @@
 import static org.exist.util.ThreadUtils.newInstanceThread;
 
 /**
- * Manages the journalling log. The database uses one central journal for
+ * Manages the journal log. The database uses one central journal for
  * all data files. If the journal exceeds the predefined maximum size, a new file is created.
  * Every journal file has a unique number, which keeps growing during the lifetime of the db.
  * The name of the file corresponds to the file number. The file with the highest
  * number will be used for recovery.
- * <p>
+ *
  * A buffer is used to temporarily buffer journal entries. To guarantee consistency, the buffer will be flushed
- * and the journal is synched after every commit or whenever a db page is written to disk.
- * <p>
- * Each entry has the structure:
+ * and the journal is synced after every commit or whenever a db page is written to disk.
+ *
+ * Each journal file has the following format:
+ *
+ * <pre>{@code
+ *     [magicNumber, version, entry*]
+ * }</pre>
+ *
+ * {@code magicNumber}  4 bytes with the value {@link #JOURNAL_MAGIC_NUMBER}.
+ * {@code version}      2 bytes (java.lang.short) with the value {@link #JOURNAL_VERSION}.
+ * {@code entry}        one or more variable length journal {@code entry} records.
+ *
+ * Each {@code entry} record has the format:
+ *
+ * <pre>{@code
+ *     [entryHeader, data, backLink, checksum]
+ * }</pre>
+ *
+ * {@code entryHeader}      11 bytes describes the entry (see below).
+ * {@code data}             {@code entryHeader->length} bytes of data for the entry.
+ * {@code backLink}         2 bytes (java.lang.short) offset to the start of the entry record, calculated by {@code entryHeader.length + dataLength}.
+ *                              The offset for the start of the entry record can be calculated as {@code endOfRecordOffset - 8 - 2 - backLink}.
+ *                              This is used when scanning the log file backwards for recovery.
+ * {@code checksum}         8 bytes for a 64 bit checksum. The checksum includes the {@code entryHeader}, {@code data}, and {@code backLink}.
  *
- * <pre>[byte: entryType, long: transactionId, short length, byte[] data, short backLink]</pre>
+ * The {@code entryHeader} has the format:
  *
- * <ul>
- * <li>entryType is a unique id that identifies the log record. Entry types are registered via the
- * {@link org.exist.storage.journal.LogEntryTypes} class.</li>
- * <li>transactionId: the id of the transaction that created the record.</li>
- * <li>length: the length of the log entry data.</li>
- * <li>data: the payload data provided by the {@link org.exist.storage.journal.Loggable} object.</li>
- * <li>backLink: offset to the start of the record. Used when scanning the log file backwards.</li>
- * </ul>
+ * <pre>{@code
+ *     [entryType, transactionId, dataLength]
+ * }</pre>
+ *
+ * {@code entryType}        1 byte indicates the type of the entry.
+ * {@code transactionId}    8 bytes (java.lang.long) the id of the transaction that created the record.
+ * {@code dataLength}       2 bytes (java.lang.short) the length of the log entry {@code data}.
  *
  * @author wolf
+ * @author aretter
  */
 @ConfigurationClass("journal")
 //TODO: conf.xml refactoring <recovery> => <recovery><journal/></recovery>
@@ -88,7 +111,7 @@ public final class Journal {
      */
     public static final int JOURNAL_HEADER_LEN = 6;
     public static final byte[] JOURNAL_MAGIC_NUMBER = {0x0E, 0x0D, 0x0B, 0x01};
-    public static final short JOURNAL_VERSION = 2;
+    public static final short JOURNAL_VERSION = 3;
 
     public static final String RECOVERY_SYNC_ON_COMMIT_ATTRIBUTE = "sync-on-commit";
     public static final String RECOVERY_JOURNAL_DIR_ATTRIBUTE = "journal-dir";
@@ -115,19 +138,35 @@ public final class Journal {
     public static final int LOG_ENTRY_BACK_LINK_LEN = 2;
 
     /**
-     * header length + trailing back link
+     * the length of the checkum in a log entry
+     */
+    public static final int LOG_ENTRY_CHECKSUM_LEN = 8;
+
+    /**
+     * header length + trailing back link length + checksum length
      */
-    public static final int LOG_ENTRY_BASE_LEN = LOG_ENTRY_HEADER_LEN + LOG_ENTRY_BACK_LINK_LEN;
+    public static final int LOG_ENTRY_BASE_LEN = LOG_ENTRY_HEADER_LEN + LOG_ENTRY_BACK_LINK_LEN + LOG_ENTRY_CHECKSUM_LEN;
 
     /**
      * default maximum journal size
      */
-    public static final int DEFAULT_MAX_SIZE = 100; //MB
+    public static final int DEFAULT_MAX_SIZE = 100;  //MB
 
     /**
      * minimal size the journal needs to have to be replaced by a new file during a checkpoint
      */
-    private static final int DEFAULT_MIN_SIZE = 1; // MB
+    private static final int DEFAULT_MIN_SIZE = 1;  // MB
+
+    /**
+     * We use a 1 megabyte buffer.
+     */
+    public static final int BUFFER_SIZE = 1024 * 1024;  // bytes
+
+    /**
+     * Seed used for xxhash-64 checksums calculated
+     * by the journal.
+     */
+    public static final long XXHASH64_SEED = 0x9747b28c;
 
     /**
      * Minimum size limit for the journal file before it is replaced by a new file.
@@ -217,11 +256,12 @@ public final class Journal {
 
     private volatile boolean initialised = false;
 
+    private final XXHash64 xxHash64 = XXHashFactory.fastestInstance().hash64();
+
     public Journal(final BrokerPool pool, final Path directory) throws EXistException {
         this.pool = pool;
         this.fsJournalDir = directory.resolve("fs.journal");
-        // we use a 1 megabyte buffer:
-        this.currentBuffer = ByteBuffer.allocateDirect(1024 * 1024);
+        this.currentBuffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
 
         this.fileSyncRunnable = new FileSyncRunnable(latch);
         this.fileSyncThread = newInstanceThread(pool, "file-sync-thread", fileSyncRunnable);
@@ -296,6 +336,11 @@ public synchronized void writeToLog(final Loggable entry) throws JournalExceptio
 
         SanityCheck.ASSERT(!inRecovery, "Write to log during recovery. Should not happen!");
         final int size = entry.getLogSize();
+
+        if (size > Short.MAX_VALUE) {
+            throw new JournalException("Journal can only write log entries of less that 32KB");
+        }
+
         final int required = size + LOG_ENTRY_BASE_LEN;
         if (required > currentBuffer.remaining()) {
             flushToLog(false);
@@ -313,11 +358,22 @@ public synchronized void writeToLog(final Loggable entry) throws JournalExceptio
         entry.setLsn(currentLsn);
 
         try {
+            final int currentBufferEntryOffset = currentBuffer.position();
+
+            // write entryHeader
             currentBuffer.put(entry.getLogType());
             currentBuffer.putLong(entry.getTransactionId());
-            currentBuffer.putShort((short) entry.getLogSize());
+            currentBuffer.putShort((short) size);
+
+            // write entry data
             entry.write(currentBuffer);
+
+            // write backlink
             currentBuffer.putShort((short) (size + LOG_ENTRY_HEADER_LEN));
+
+            // write checksum
+            final long checksum = xxHash64.hash(currentBuffer, currentBufferEntryOffset, currentBuffer.position() - currentBufferEntryOffset, XXHASH64_SEED);
+            currentBuffer.putLong(checksum);
         } catch (final BufferOverflowException e) {
             throw new JournalException("Buffer overflow while writing log record: " + entry.dump(), e);
         }
@@ -499,7 +555,7 @@ public void switchFiles() throws LogException {
     }
 
     private void writeJournalHeader(final SeekableByteChannel channel) throws IOException {
-        final ByteBuffer buf = ByteBuffer.allocate(JOURNAL_HEADER_LEN);
+        final ByteBuffer buf = ByteBuffer.allocateDirect(JOURNAL_HEADER_LEN);
 
         // write the magic number
         buf.put(JOURNAL_MAGIC_NUMBER);

src/org/exist/storage/journal/Journal.java

@@ -19,6 +19,8 @@
  */
 package org.exist.storage.journal;
 
+import net.jpountz.xxhash.StreamingXXHash64;
+import net.jpountz.xxhash.XXHashFactory;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.exist.storage.DBBroker;
@@ -52,6 +54,8 @@ public class JournalReader implements AutoCloseable {
     @Nullable
     private SeekableByteChannel fc;
 
+    private final StreamingXXHash64 xxHash64 = XXHashFactory.fastestInstance().newStreamingHash64(Journal.XXHASH64_SEED);
+
     /**
      * Opens the specified file for reading.
      *
@@ -74,7 +78,7 @@ public JournalReader(final DBBroker broker, final Path file, final int fileNumbe
 
     private void validateJournalHeader(final Path file, final SeekableByteChannel fc) throws IOException, LogException {
         // read the magic number
-        final ByteBuffer buf = ByteBuffer.allocate(JOURNAL_HEADER_LEN);
+        final ByteBuffer buf = ByteBuffer.allocateDirect(JOURNAL_HEADER_LEN);
         fc.read(buf);
         buf.flip();
 
@@ -137,18 +141,18 @@ Loggable previousEntry() throws LogException {
                 return null;
             }
 
-            // go back two bytes and read the back-link of the last entry
-            fc.position(fc.position() - LOG_ENTRY_BACK_LINK_LEN);
+            // go back 8 bytes (checksum length) + 2 bytes (backLink length) and read the backLink (2 bytes) of the last entry
+            fc.position(fc.position() - LOG_ENTRY_CHECKSUM_LEN - LOG_ENTRY_BACK_LINK_LEN);
             header.clear().limit(LOG_ENTRY_BACK_LINK_LEN);
             final int read = fc.read(header);
             if (read != LOG_ENTRY_BACK_LINK_LEN) {
                 throw new LogException("Unable to read journal entry back-link!");
             }
             header.flip();
-            final short prevLink = header.getShort();
+            final short backLink = header.getShort();
 
             // position the channel to the start of the previous entry and mark it
-            final long prevStart = fc.position() - LOG_ENTRY_BACK_LINK_LEN - prevLink;
+            final long prevStart = fc.position() - LOG_ENTRY_BACK_LINK_LEN - backLink;
             fc.position(prevStart);
             final Loggable loggable = readEntry();
 
@@ -204,6 +208,19 @@ Loggable readEntry() throws LogException {
             }
             header.flip();
 
+            // prepare the checksum for the header
+            xxHash64.reset();
+            if (header.hasArray()) {
+                xxHash64.update(header.array(), 0, LOG_ENTRY_HEADER_LEN);
+            } else {
+                final int mark = header.position();
+                header.position(0);
+                final byte buf[] = new byte[LOG_ENTRY_HEADER_LEN];
+                header.get(buf);
+                xxHash64.update(buf, 0, LOG_ENTRY_HEADER_LEN);
+                header.position(mark);
+            }
+
             final byte entryType = header.get();
             final long transactId = header.getLong();
             final short size = header.getShort();
@@ -218,23 +235,51 @@ Loggable readEntry() throws LogException {
             }
             loggable.setLsn(lsn);
 
-            if (size + LOG_ENTRY_BACK_LINK_LEN > payload.capacity()) {
+            final int remainingEntryBytes = size + LOG_ENTRY_BACK_LINK_LEN + LOG_ENTRY_CHECKSUM_LEN;
+
+            if (remainingEntryBytes > payload.capacity()) {
                 // resize the payload buffer
-                payload = ByteBuffer.allocate(size + LOG_ENTRY_BACK_LINK_LEN);
+                payload = ByteBuffer.allocateDirect(remainingEntryBytes);
             }
-            payload.clear().limit(size + LOG_ENTRY_BACK_LINK_LEN);
+            payload.clear().limit(remainingEntryBytes);
             read = fc.read(payload);
-            if (read < size + LOG_ENTRY_BACK_LINK_LEN) {
+            if (read < remainingEntryBytes) {
                 throw new LogException("Incomplete log entry found!");
             }
             payload.flip();
+
+            // read entry data
             loggable.read(payload);
-            final short prevLink = payload.getShort();
-            if (prevLink != size + LOG_ENTRY_HEADER_LEN) {
-                LOG.error("Bad pointer to previous: prevLink = " + prevLink + "; size = " + size +
+
+            // read entry backLink
+            final short backLink = payload.getShort();
+            if (backLink != size + LOG_ENTRY_HEADER_LEN) {
+                LOG.error("Bad pointer to previous: backLink = " + backLink + "; size = " + size +
                         "; transactId = " + transactId);
                 throw new LogException("Bad pointer to previous in entry: " + loggable.dump());
             }
+
+            // update the checksum for the entry data and backLink
+            if (payload.hasArray()) {
+                xxHash64.update(payload.array(), 0, size + LOG_ENTRY_BACK_LINK_LEN);
+            } else {
+                final int mark = payload.position();
+                payload.position(0);
+                final byte buf[] = new byte[size + LOG_ENTRY_BACK_LINK_LEN];
+                payload.get(buf);
+                xxHash64.update(buf, 0, size + LOG_ENTRY_BACK_LINK_LEN);
+                payload.position(mark);
+            }
+
+            // read the entry checksum
+            final long checksum = payload.getLong();
+
+            // verify the checksum
+            final long calculatedChecksum = xxHash64.getValue();
+            if (checksum != calculatedChecksum) {
+                throw new LogException("Checksum mismatch whilst reading log entry. read=" + checksum + " calculated=" + calculatedChecksum);
+            }
+
             return loggable;
         } catch (final IOException e) {
             throw new LogException(e.getMessage(), e);