improve compactionProgress stats

Author: staltz

README.md

@@ -178,6 +178,45 @@ log.since((offset) => {
 })
 ```
 
+### Compact the log (remove deleted records)
+
+```js
+log.compact((err) => {
+  // This callback will be called once, when the compaction is done.
+})
+```
+
+### Track progress of compactions
+
+As an [obz] observable:
+
+```js
+log.compactionProgress((progress) => {
+  console.log(progress)
+  // {
+  //   startOffset,
+  //   compactedOffset,
+  //   unshiftedOffset,
+  //   percent,
+  //   done,
+  //   sizeDiff,
+  // }
+})
+```
+
+Where
+
+- `startOffset`: the starting point for compaction. All offsets smaller than
+  this have been left untouched by the compaction algorithm.
+- `compactedOffset`: all records up until this point have been compacted so far.
+- `unshiftedOffset`: offset for the first record that hasn't yet been "moved"
+  to previous slots. Tracking this allows you to see the algorithm proceeding.
+- `percent`: a number between 0 and 1 to indicate the progress of compaction.
+- `done`: a boolean indicating whether compaction is ongoing (`false`) or done
+  (`true`).
+- `sizeDiff`: number of bytes freed after compaction is finished. Only available
+  if `done` is `true`.
+
 ### Close the log
 
 ```js

compaction.js

@@ -118,7 +118,8 @@ function PersistentState(logFilename, blockSize) {
  */
 function Compaction(log, onDone) {
   const persistentState = PersistentState(log.filename, log.blockSize)
-  const since = Obv() // for the unshifted offset
+  const progress = Obv() // for the unshifted offset
+  let startOffset = 0
 
   let compactedBlockIndex = -1
   let compactedBlockBuf = null
@@ -131,6 +132,7 @@ function Compaction(log, onDone) {
 
   loadPersistentState(function onCompactionStateLoaded2(err) {
     if (err) return onDone(err)
+    startOffset = compactedBlockIndex * log.blockSize
     compactedBlockIndex -= 1 // because it'll be incremented very soon
     compactNextBlock()
   })
@@ -341,17 +343,28 @@ function Compaction(log, onDone) {
   }
 
   function compactNextBlock() {
-    since.set(unshiftedOffset)
     compactedBlockIndex += 1
     compactedBlockBuf = Buffer.alloc(log.blockSize)
     compactedOffset = compactedBlockIndex * log.blockSize
     compactedBlockIdenticalToUnshifted = true
+    progress.set(calculateProgressStats())
     savePersistentState(function onCompactionStateSaved(err) {
       if (err) return onDone(err)
       continueCompactingBlock()
     })
   }
 
+  function calculateProgressStats() {
+    const percent =
+      (unshiftedOffset - startOffset) / (log.since.value - startOffset)
+    return {
+      startOffset,
+      compactedOffset,
+      unshiftedOffset,
+      percent,
+    }
+  }
+
   function stop() {
     compactedBlockBuf = null
     unshiftedBlockBuf = null
@@ -362,7 +375,7 @@ function Compaction(log, onDone) {
   }
 
   return {
-    since,
+    progress,
   }
 }
 

index.js

@@ -60,7 +60,7 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
   let nextOffsetInBlock = null
   const since = Obv() // offset of last written record
   let compaction = null
-  const compactionSince = Obv()
+  const compactionProgress = Obv()
   const waitingCompaction = []
 
   onLoad(function maybeResumeCompaction() {
@@ -442,12 +442,12 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
       compaction = new Compaction(self, (err, sizeDiff) => {
         compaction = null
         if (err) return cb(err)
-        compactionSince.set({ value: sizeDiff, done: true })
+        compactionProgress.set({ sizeDiff, percent: 1, done: true })
         while (waitingCompaction.length) waitingCompaction.shift()()
         cb()
       })
-      compaction.since((unshiftedOffset) => {
-        compactionSince.set({ value: unshiftedOffset, done: false })
+      compaction.progress((stats) => {
+        compactionProgress.set({ ...stats, done: false })
       })
     })
   }
@@ -499,7 +499,7 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
     onDrain: onLoad(onDrain),
     compact: onLoad(compact),
     since,
-    compactionSince,
+    compactionProgress,
     stream(opts) {
       const stream = new Stream(self, opts)
       self.streams.push(stream)

test/compaction.js

@@ -178,14 +178,47 @@ tape('shift many blocks', async (t) => {
   })
 
   const progressArr = []
-  log.compactionSince((obj) => {
-    progressArr.push(obj)
+  log.compactionProgress((stats) => {
+    progressArr.push(stats)
   })
 
   const [err] = await run(log.compact)()
   await run(log.onDrain)()
   t.error(err, 'no error when compacting')
 
+  t.deepEquals(
+    progressArr,
+    [
+      {
+        startOffset: 11,
+        compactedOffset: 11,
+        unshiftedOffset: 11,
+        percent: 0,
+        done: false,
+      },
+      {
+        startOffset: 11,
+        compactedOffset: 22,
+        unshiftedOffset: 28,
+        percent: 0.4358974358974359,
+        done: false,
+      },
+      {
+        startOffset: 11,
+        compactedOffset: 33,
+        unshiftedOffset: 44,
+        percent: 0.8461538461538461,
+        done: false,
+      },
+      {
+        sizeDiff: 11, // the log is now 1 block shorter
+        percent: 1,
+        done: true,
+      },
+    ],
+    'progress events'
+  )
+
   await new Promise((resolve) => {
     log.stream({ offsets: false }).pipe(
       push.collect((err, ary) => {
@@ -209,17 +242,6 @@ tape('shift many blocks', async (t) => {
     )
   })
 
-  t.deepEquals(
-    progressArr,
-    [
-      { value: 11 + 0, done: false },
-      { value: 22 + 6, done: false },
-      { value: 44 + 0, done: false },
-      { value: 11, done: true }, // the log is now 1 block shorter
-    ],
-    'progress events'
-  )
-
   await run(log.close)()
   t.end()
 })