Merge pull request #70 from ssb-ngi-pointer/compact-reindex

first compactionProgress events should have startOffset

Author: staltz

compaction.js

@@ -25,23 +25,25 @@ const NO_TRUNCATE = 0xffffffff
  *
  * - 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
+ * - bytes 4..7: UInt32LE for the startOffset, usually the start of some block
+ * - bytes 8..11: UInt32LE for block index where to perform truncation
+ *                where 0xFFFFFFFF means no truncation to-be-done yet
+ * - bytes 12..15: UInt32LE for the blockIndex to-be-compacted
+ * - bytes 16..19: UInt32LE for the 1st unshifted record's offset
+ * - bytes 20..(20+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
+  const stateFileSize = 4 + 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,
+          startOffset: 0,
           truncateBlockIndex: NO_TRUNCATE,
           compactedBlockIndex: 0,
           unshiftedOffset: 0,
@@ -54,10 +56,11 @@ function PersistentState(logFilename, blockSize) {
           if (err) return cb(err)
           const state = {
             version: buf.readUInt32LE(0),
-            truncateBlockIndex: buf.readUInt32LE(4),
-            compactedBlockIndex: buf.readUInt32LE(8),
-            unshiftedOffset: buf.readUInt32LE(12),
-            unshiftedBlockBuf: buf.slice(16),
+            startOffset: buf.readUInt32LE(4),
+            truncateBlockIndex: buf.readUInt32LE(8),
+            compactedBlockIndex: buf.readUInt32LE(12),
+            unshiftedOffset: buf.readUInt32LE(16),
+            unshiftedBlockBuf: buf.slice(20),
             initial: false,
           }
           cb(null, state)
@@ -69,10 +72,11 @@ function PersistentState(logFilename, blockSize) {
   function save(state, cb) {
     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)
+    buf.writeUint32LE(state.startOffset, 4)
+    buf.writeUInt32LE(state.truncateBlockIndex, 8)
+    buf.writeUInt32LE(state.compactedBlockIndex, 12)
+    buf.writeUInt32LE(state.unshiftedOffset, 16)
+    state.unshiftedBlockBuf.copy(buf, 20)
     writeLock((unlock) => {
       raf.write(0, buf, function onRafWriteDone(err) {
         if (err) return unlock(cb, err)
@@ -147,7 +151,6 @@ function Compaction(log, onDone) {
     if (truncateBlockIndex !== NO_TRUNCATE) {
       truncateAndBeDone()
     } else {
-      startOffset = compactedBlockIndex * log.blockSize
       compactedBlockIndex -= 1 // because it'll be incremented very soon
       compactNextBlock()
     }
@@ -158,6 +161,7 @@ function Compaction(log, onDone) {
       if (err) return cb(err)
       if (state.version !== 1) return cb(new Error('unsupported state version'))
       version = state.version
+      startOffset = state.startOffset
       truncateBlockIndex = state.truncateBlockIndex
       compactedBlockIndex = state.compactedBlockIndex
       unshiftedOffset = state.unshiftedOffset
@@ -166,6 +170,7 @@ function Compaction(log, onDone) {
       if (state.initial) {
         findStateFromLog(function foundStateFromLog(err, state) {
           if (err) return cb(err)
+          startOffset = state.compactedBlockIndex * log.blockSize
           compactedBlockIndex = state.compactedBlockIndex
           unshiftedOffset = state.unshiftedOffset
           unshiftedBlockBuf = state.unshiftedBlockBuf
@@ -191,6 +196,7 @@ function Compaction(log, onDone) {
       persistentState.save(
         {
           version,
+          startOffset,
           truncateBlockIndex,
           compactedBlockIndex,
           unshiftedOffset,
@@ -388,6 +394,7 @@ function Compaction(log, onDone) {
     truncateBlockIndex = compactedBlockIndex
     const state = {
       version,
+      startOffset,
       truncateBlockIndex,
       compactedBlockIndex: 0,
       unshiftedOffset: 0,

index.js

@@ -62,7 +62,11 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
   let nextOffsetInBlock = null
   const since = Obv() // offset of last written record
   let compaction = null
-  const compactionProgress = Obv()
+  const compactionProgress = Obv().set(
+    Compaction.stateFileExists(filename)
+      ? { done: false }
+      : { sizeDiff: 0, percent: 1, done: true }
+  )
   const waitingCompaction = []
 
   onLoad(function maybeResumeCompaction() {

test/compaction.js

@@ -189,6 +189,11 @@ tape('shift many blocks', async (t) => {
   t.deepEquals(
     progressArr,
     [
+      {
+        sizeDiff: 0,
+        percent: 1,
+        done: true,
+      },
       {
         startOffset: 11,
         compactedOffset: 11,
@@ -463,6 +468,88 @@ tape('compact handling holes of different sizes', async (t) => {
   t.end()
 })
 
+tape('startOffset is correct', async (t) => {
+  t.timeoutAfter(6000)
+  const file = '/tmp/compaction-test_' + Date.now() + '.log'
+  const log = Log(file, { blockSize: 9, codec: hexCodec })
+
+  await run(log.append)(
+    [
+      // block 0
+      [0x11, 0x22], // offsets: 0, 3
+      // block 1
+      [0x33, 0x44], // offsets: 9+0, 9+3
+    ].flat()
+  )
+  await run(log.onDrain)()
+  t.pass('append four records')
+
+  await run(log.del)(0)
+  await run(log.onDeletesFlushed)()
+  t.pass('delete 1st record')
+
+  const progressArr = []
+  log.compactionProgress((stats) => {
+    progressArr.push(stats)
+  })
+
+  const [err] = await run(log.compact)()
+  t.error(err, 'no error when compacting')
+
+  t.deepEquals(
+    progressArr,
+    [
+      {
+        sizeDiff: 0,
+        percent: 1,
+        done: true,
+      },
+      {
+        startOffset: 0,
+        compactedOffset: 0,
+        unshiftedOffset: 3,
+        percent: 0.25,
+        done: false,
+      },
+      {
+        startOffset: 0,
+        compactedOffset: 9,
+        unshiftedOffset: 12,
+        percent: 1,
+        done: false,
+      },
+      {
+        sizeDiff: 0,
+        percent: 1,
+        done: true,
+      },
+    ],
+    'progress events'
+  )
+
+  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, 0x33],
+            // block 1
+            [0x44],
+          ].flat(),
+          'log has 2 blocks'
+        )
+        resolve()
+      })
+    )
+  })
+
+  await run(log.close)()
+  t.end()
+})
+
 tape('recovers from crash just after persisting state', async (t) => {
   t.timeoutAfter(6000)
   const file = '/tmp/compaction-test_' + Date.now() + '.log'
@@ -482,6 +569,7 @@ tape('recovers from crash just after persisting state', async (t) => {
   t.pass('suppose compaction was in progress: [0x22, 0x33] and [0x33, 0x44]')
 
   const version = [1, 0, 0, 0] // uint32LE
+  const startOffset = [0, 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
@@ -494,6 +582,7 @@ tape('recovers from crash just after persisting state', async (t) => {
     file + '.compaction',
     Buffer.from([
       ...version,
+      ...startOffset,
       ...truncateBlockIndex,
       ...compactingBlockIndex,
       ...unshiftedOffset,
@@ -506,8 +595,35 @@ tape('recovers from crash just after persisting state', async (t) => {
   log = Log(file, { blockSize: 9, codec: hexCodec })
   t.pass('start log, compaction should autostart')
 
+  const progressArr = []
+  log.compactionProgress((stats) => {
+    progressArr.push(stats)
+  })
+
   await timer(1000)
 
+  t.deepEquals(
+    progressArr,
+    [
+      {
+        done: false,
+      },
+      {
+        startOffset: 0,
+        compactedOffset: 9,
+        unshiftedOffset: 12,
+        percent: 1,
+        done: false,
+      },
+      {
+        sizeDiff: 0,
+        percent: 1,
+        done: true,
+      },
+    ],
+    'progress events'
+  )
+
   await new Promise((resolve) => {
     log.stream({ offsets: false }).pipe(
       push.collect((err, ary) => {
@@ -552,6 +668,7 @@ tape('recovers from crash just after persisting block', async (t) => {
   t.pass('suppose compaction was in progress: [0x22, 0x33] and [0x33, 0x44]')
 
   const version = [1, 0, 0, 0] // uint32LE
+  const startOffset = [0,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
@@ -564,6 +681,7 @@ tape('recovers from crash just after persisting block', async (t) => {
     file + '.compaction',
     Buffer.from([
       ...version,
+      ...startOffset,
       ...truncateBlockIndex,
       ...compactingBlockIndex,
       ...unshiftedOffset,
@@ -621,6 +739,7 @@ tape('restarts from crash just before truncating log', async (t) => {
   t.pass('suppose compaction ready: [0x22, 0x44], [0x55, 0x66], [0x55, 0x66]')
 
   const version = [1, 0, 0, 0] // uint32LE
+  const startOffset = [0,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
@@ -629,6 +748,7 @@ tape('restarts from crash just before truncating log', async (t) => {
     file + '.compaction',
     Buffer.from([
       ...version,
+      ...startOffset,
       ...truncateBlockIndex,
       ...compactingBlockIndex,
       ...unshiftedOffset,