Merge pull request #69 from ssb-ngi-pointer/crash-before-truncate

Crash before truncate

Author: staltz

compaction.js

@@ -18,36 +18,46 @@ function stateFileExists(logFilename) {
   return fs.existsSync(getStateFilename(logFilename))
 }
 
+const NO_TRUNCATE = 0xffffffff
+
 /**
  * This file has state describing the continuation of the compaction algorithm.
  *
- * - bytes 0..3: UInt32LE for the blockIndex to-be-compacted
- * - bytes 4..7: UInt32LE for the 1st unshifted record's offset
- * - bytes 8..(8+blockSize-1): blockBuf containing the 1st unshifted record
+ * - bytes 0..3: UInt32LE for the version of this file format
+ *               smallest version is 1.
+ * - bytes 4..7: UInt32LE for block index where to perform truncation
+ *               where 0xFFFFFFFF means no truncation to-be-done yet
+ * - bytes 8..11: UInt32LE for the blockIndex to-be-compacted
+ * - bytes 12..15: UInt32LE for the 1st unshifted record's offset
+ * - bytes 16..(16+blockSize-1): blockBuf containing the 1st unshifted record
  */
 function PersistentState(logFilename, blockSize) {
   const raf = RAF(getStateFilename(logFilename))
   const writeLock = mutexify()
+  const stateFileSize = 4 + 4 + 4 + 4 + blockSize
 
   function load(cb) {
     raf.stat(function onRAFStatDone(err, stat) {
       const fileSize = !err && stat ? stat.size : -1
       if (fileSize <= 0) {
         const state = {
+          version: 1,
+          truncateBlockIndex: NO_TRUNCATE,
           compactedBlockIndex: 0,
           unshiftedOffset: 0,
           unshiftedBlockBuffer: null,
           initial: true,
         }
         cb(null, state)
       } else {
-        const stateFileSize = 4 + 4 + blockSize
         raf.read(0, stateFileSize, function onFirstRAFReadDone(err, buf) {
           if (err) return cb(err)
           const state = {
-            compactedBlockIndex: buf.readUInt32LE(0),
-            unshiftedOffset: buf.readUInt32LE(4),
-            unshiftedBlockBuf: buf.slice(8),
+            version: buf.readUInt32LE(0),
+            truncateBlockIndex: buf.readUInt32LE(4),
+            compactedBlockIndex: buf.readUInt32LE(8),
+            unshiftedOffset: buf.readUInt32LE(12),
+            unshiftedBlockBuf: buf.slice(16),
             initial: false,
           }
           cb(null, state)
@@ -57,10 +67,12 @@ function PersistentState(logFilename, blockSize) {
   }
 
   function save(state, cb) {
-    const buf = Buffer.alloc(4 + 4 + blockSize)
-    buf.writeUInt32LE(state.compactedBlockIndex, 0)
-    buf.writeUInt32LE(state.unshiftedOffset, 4)
-    state.unshiftedBlockBuf.copy(buf, 8)
+    const buf = Buffer.alloc(stateFileSize)
+    buf.writeUInt32LE(state.version, 0)
+    buf.writeUInt32LE(state.truncateBlockIndex, 4)
+    buf.writeUInt32LE(state.compactedBlockIndex, 8)
+    buf.writeUInt32LE(state.unshiftedOffset, 12)
+    state.unshiftedBlockBuf.copy(buf, 16)
     writeLock((unlock) => {
       raf.write(0, buf, function onRafWriteDone(err) {
         if (err) return unlock(cb, err)
@@ -78,10 +90,7 @@ function PersistentState(logFilename, blockSize) {
     if (stateFileExists(logFilename)) {
       raf.close(function onRAFClosed(err) {
         if (err) return cb(err)
-        fs.unlink(raf.filename, function onStateFileDeleted(err) {
-          if (err) return cb(err)
-          else cb()
-        })
+        fs.unlink(raf.filename, cb)
       })
     } else {
       cb()
@@ -120,6 +129,7 @@ function Compaction(log, onDone) {
   const persistentState = PersistentState(log.filename, log.blockSize)
   const progress = Obv() // for the unshifted offset
   let startOffset = 0
+  let version = 0
 
   let compactedBlockIndex = -1
   let compactedBlockBuf = null
@@ -130,16 +140,25 @@ function Compaction(log, onDone) {
   let unshiftedBlockBuf = null
   let unshiftedOffset = 0
 
+  let truncateBlockIndex = NO_TRUNCATE
+
   loadPersistentState(function onCompactionStateLoaded2(err) {
     if (err) return onDone(err)
-    startOffset = compactedBlockIndex * log.blockSize
-    compactedBlockIndex -= 1 // because it'll be incremented very soon
-    compactNextBlock()
+    if (truncateBlockIndex !== NO_TRUNCATE) {
+      truncateAndBeDone()
+    } else {
+      startOffset = compactedBlockIndex * log.blockSize
+      compactedBlockIndex -= 1 // because it'll be incremented very soon
+      compactNextBlock()
+    }
   })
 
   function loadPersistentState(cb) {
     persistentState.load(function onCompactionStateLoaded(err, state) {
       if (err) return cb(err)
+      if (state.version !== 1) return cb(new Error('unsupported state version'))
+      version = state.version
+      truncateBlockIndex = state.truncateBlockIndex
       compactedBlockIndex = state.compactedBlockIndex
       unshiftedOffset = state.unshiftedOffset
       unshiftedBlockBuf = state.unshiftedBlockBuf
@@ -163,16 +182,16 @@ function Compaction(log, onDone) {
 
   function savePersistentState(cb) {
     if (!unshiftedBlockBuf) {
-      loadUnshiftedBlock(function onUnshiftedBlockLoaded() {
-        saveIt.call(this)
-      })
+      loadUnshiftedBlock(saveIt)
     } else {
-      saveIt.call(this)
+      saveIt()
     }
 
     function saveIt() {
       persistentState.save(
         {
+          version,
+          truncateBlockIndex,
           compactedBlockIndex,
           unshiftedOffset,
           unshiftedBlockBuf,
@@ -249,9 +268,7 @@ function Compaction(log, onDone) {
     while (true) {
       // Fetch the unshifted block, if necessary
       if (!unshiftedBlockBuf) {
-        loadUnshiftedBlock(function onUnshiftedBlockLoaded() {
-          continueCompactingBlock()
-        })
+        loadUnshiftedBlock(continueCompactingBlock)
         return
       }
       // When all records have been shifted (thus end of log), stop compacting
@@ -273,7 +290,7 @@ function Compaction(log, onDone) {
       // Proceed to compact the next block if this block is full
       if (log.hasNoSpaceFor(unshiftedDataBuf, offsetInCompactedBlock)) {
         saveCompactedBlock()
-        setImmediate(() => compactNextBlock())
+        setImmediate(compactNextBlock)
         return
       }
 
@@ -368,9 +385,30 @@ function Compaction(log, onDone) {
   function stop() {
     compactedBlockBuf = null
     unshiftedBlockBuf = null
-    persistentState.destroy(function onCompactionStateDestroyed(err) {
+    truncateBlockIndex = compactedBlockIndex
+    const state = {
+      version,
+      truncateBlockIndex,
+      compactedBlockIndex: 0,
+      unshiftedOffset: 0,
+      unshiftedBlockBuf: Buffer.alloc(0),
+    }
+    persistentState.save(state, function onTruncateStateSaved(err) {
       if (err) return onDone(err)
-      log.truncate(compactedBlockIndex, onDone)
+      truncateAndBeDone()
+    })
+  }
+
+  function truncateAndBeDone() {
+    if (truncateAndBeDone === NO_TRUNCATE) {
+      return onDone(new Error('Cannot truncate log yet'))
+    }
+    log.truncate(truncateBlockIndex, function onTruncatedLog(err, sizeDiff) {
+      if (err) return onDone(err)
+      persistentState.destroy(function onStateDestroyed(err) {
+        if (err) return onDone(err)
+        onDone(null, sizeDiff)
+      })
     })
   }
 

test/compaction.js

@@ -481,6 +481,8 @@ tape('recovers from crash just after persisting state', async (t) => {
   await run(log.close)()
   t.pass('suppose compaction was in progress: [0x22, 0x33] and [0x33, 0x44]')
 
+  const version = [1, 0, 0, 0] // uint32LE
+  const truncateBlockIndex = [255, 255, 255, 255] //uint32LE
   const compactingBlockIndex = [1, 0, 0, 0] // uint32LE
   const unshiftedOffset = [9 + 3, 0, 0, 0] // uint32LE
   const unshiftedBlock = [
@@ -491,6 +493,8 @@ tape('recovers from crash just after persisting state', async (t) => {
   await fs.promises.writeFile(
     file + '.compaction',
     Buffer.from([
+      ...version,
+      ...truncateBlockIndex,
       ...compactingBlockIndex,
       ...unshiftedOffset,
       ...unshiftedBlock,
@@ -547,6 +551,8 @@ tape('recovers from crash just after persisting block', async (t) => {
   await run(log.close)()
   t.pass('suppose compaction was in progress: [0x22, 0x33] and [0x33, 0x44]')
 
+  const version = [1, 0, 0, 0] // uint32LE
+  const truncateBlockIndex = [255, 255, 255, 255] //uint32LE
   const compactingBlockIndex = [0, 0, 0, 0] // uint32LE
   const unshiftedOffset = [0, 0, 0, 0] // uint32LE
   const unshiftedBlock = [
@@ -557,6 +563,8 @@ tape('recovers from crash just after persisting block', async (t) => {
   await fs.promises.writeFile(
     file + '.compaction',
     Buffer.from([
+      ...version,
+      ...truncateBlockIndex,
       ...compactingBlockIndex,
       ...unshiftedOffset,
       ...unshiftedBlock,
@@ -592,6 +600,71 @@ tape('recovers from crash just after persisting block', async (t) => {
   t.end()
 })
 
+tape('restarts from crash just before truncating log', async (t) => {
+  t.timeoutAfter(6000)
+  const file = '/tmp/compaction-test_' + Date.now() + '.log'
+  let log = Log(file, { blockSize: 9, codec: hexCodec })
+  t.pass('suppose the log has blockSize 9')
+  t.pass('suppose we had blocks: [null, 0x22], [null, 0x44] and [0x55, 0x66]')
+
+  await run(log.append)(
+    [
+      // block 0
+      [0x22, 0x44], // offsets: 0, 3
+      // block 1
+      [0x55, 0x66], // offsets: 9+0, 9+3
+      // block 2
+      [0x55, 0x66], // offsets: 18+0, 18+3
+    ].flat()
+  )
+  await run(log.close)()
+  t.pass('suppose compaction ready: [0x22, 0x44], [0x55, 0x66], [0x55, 0x66]')
+
+  const version = [1, 0, 0, 0] // uint32LE
+  const truncateBlockIndex = [1, 0, 0, 0] //uint32LE
+  const compactingBlockIndex = [0, 0, 0, 0] // uint32LE
+  const unshiftedOffset = [0, 0, 0, 0] // uint32LE
+  const unshiftedBlock = [0, 0, 0, 0, 0, 0, 0, 0, 0]
+  await fs.promises.writeFile(
+    file + '.compaction',
+    Buffer.from([
+      ...version,
+      ...truncateBlockIndex,
+      ...compactingBlockIndex,
+      ...unshiftedOffset,
+      ...unshiftedBlock,
+    ])
+  )
+  t.pass('suppose compaction file: truncateBlockIndex 1')
+
+  log = Log(file, { blockSize: 9, codec: hexCodec })
+  t.pass('start log, compaction should autostart')
+
+  await timer(1000)
+
+  await new Promise((resolve) => {
+    log.stream({ offsets: false }).pipe(
+      push.collect((err, ary) => {
+        t.error(err, 'no error when streaming compacted log')
+        t.deepEqual(
+          ary,
+          [
+            // block 0
+            [0x22, 0x44],
+            // block 1
+            [0x55, 0x66],
+          ].flat(),
+          'truncated to: [0x22, 0x44], [0x55, 0x66]'
+        )
+        resolve()
+      })
+    )
+  })
+
+  await run(log.close)()
+  t.end()
+})
+
 tape('append during compaction is postponed', async (t) => {
   const file = '/tmp/compaction-test_' + Date.now() + '.log'
   const log = Log(file, { blockSize: 10 })