keep state file to signal reindex post-compaction

Author: staltz

files.js

@@ -206,6 +206,58 @@ function safeFilename(filename) {
   return sanitize(result)
 }
 
+const EmptyFile = {
+  create(filename) {
+    if (typeof window !== 'undefined') {
+      // browser
+      const IdbKvStore = require('idb-kv-store')
+      const store = new IdbKvStore(filename, { disableBroadcast: true })
+      store.set('x', 'y', () => {})
+    } else {
+      // node.js
+      const fs = require('fs')
+      fs.closeSync(fs.openSync(filename, 'w'))
+    }
+  },
+
+  exists(filename, cb) {
+    if (typeof window !== 'undefined') {
+      // browser
+      const IdbKvStore = require('idb-kv-store')
+      const store = new IdbKvStore(filename, { disableBroadcast: true })
+      store.get('x', (err, y) => {
+        if (err) return cb(null, false)
+        cb(null, y === 'y')
+      })
+    } else {
+      // node.js
+      const fs = require('fs')
+      cb(null, fs.existsSync(filename))
+    }
+  },
+
+  delete(filename, cb) {
+    EmptyFile.exists(filename, (err, exists) => {
+      if (cb) return cb(err)
+      if (!exists) cb(null)
+      else EmptyFile._actuallyDelete(filename, cb)
+    })
+  },
+
+  _actuallyDelete(filename, cb) {
+    if (typeof window !== 'undefined') {
+      // browser
+      const IdbKvStore = require('idb-kv-store')
+      const store = new IdbKvStore(filename, { disableBroadcast: true })
+      store.remove('x', cb)
+    } else {
+      // node.js
+      const rimraf = require('rimraf')
+      cb(null, rimraf.sync(filename))
+    }
+  },
+}
+
 module.exports = {
   saveTypedArrayFile,
   loadTypedArrayFile,
@@ -214,5 +266,6 @@ module.exports = {
   saveBitsetFile,
   loadBitsetFile,
   listFiles,
+  EmptyFile,
   safeFilename,
 }

index.js

@@ -26,6 +26,7 @@ const {
   loadBitsetFile,
   safeFilename,
   listFiles,
+  EmptyFile,
 } = require('./files')
 
 module.exports = function (log, indexesPath) {
@@ -41,6 +42,9 @@ module.exports = function (log, indexesPath) {
   const indexes = {}
   let isReady = false
   let waiting = []
+  let compacting = false
+  let compactStartOffset = null
+  const postCompactReindexPath = path.join(indexesPath, 'post-compact-reindex')
   const waitingCompaction = []
   const coreIndexNames = ['seq', 'timestamp', 'sequence']
   const indexingActive = Obv().set(0)
@@ -87,9 +91,40 @@ module.exports = function (log, indexesPath) {
   }
 
   log.compactionProgress((stats) => {
-    if (stats.done && waitingCompaction.length > 0) {
-      for (const cb of waitingCompaction) cb(stats.holesFound)
-      waitingCompaction.length = 0
+    if (typeof stats.startOffset === 'number' && compactStartOffset === null) {
+      compactStartOffset = stats.startOffset
+    }
+
+    if (!stats.done && !compacting) {
+      compacting = true
+      EmptyFile.create(postCompactReindexPath)
+    } else if (stats.done && compacting) {
+      compacting = false
+      const offset = compactStartOffset || 0
+      compactStartOffset = null
+
+      if (stats.sizeDiff > 0) {
+        EmptyFile.exists(postCompactReindexPath, (err, exists) => {
+          if (exists) {
+            reindex(offset, (err) => {
+              if (err) console.error('reindex jitdb after compact', err)
+              conclude()
+            })
+          } else {
+            conclude()
+          }
+        })
+      } else {
+        conclude()
+      }
+
+      function conclude() {
+        EmptyFile.delete(postCompactReindexPath, () => {
+          if (waitingCompaction.length === 0) return
+          for (const cb of waitingCompaction) cb(stats.holesFound)
+          waitingCompaction.length = 0
+        })
+      }
     }
   })
 
@@ -1519,6 +1554,8 @@ module.exports = function (log, indexesPath) {
           }
         }
 
+        if (index.bitset) index.bitset.removeRange(seq, Infinity)
+
         index.offset = prevOffset
       }
     }

package.json

@@ -14,7 +14,7 @@
     "!example.js"
   ],
   "dependencies": {
-    "atomic-file-rw": "^0.2.1",
+    "atomic-file-rw": "^0.3.0",
     "binary-search-bounds": "^2.0.4",
     "bipf": "^1.6.2",
     "crc": "3.6.0",
@@ -33,16 +33,17 @@
     "pull-stream": "^3.6.14",
     "push-stream": "^11.0.0",
     "push-stream-to-pull-stream": "^1.0.3",
+    "rimraf": "^3.0.2",
     "sanitize-filename": "^1.6.3",
     "traverse": "^0.6.6",
     "typedarray-to-buffer": "^4.0.0",
     "typedfastbitset": "~0.2.1"
   },
   "peerDependencies": {
-    "async-append-only-log": "^4.2.1"
+    "async-append-only-log": "^4.3.2"
   },
   "devDependencies": {
-    "async-append-only-log": "^4.3.1",
+    "async-append-only-log": "^4.3.2",
     "expose-gc": "^1.0.0",
     "flumecodec": "0.0.1",
     "flumelog-offset": "3.4.4",
@@ -53,7 +54,6 @@
     "pretty-bytes": "^5.6.0",
     "pretty-quick": "^3.1.0",
     "pull-pushable": "^2.2.0",
-    "rimraf": "^3.0.2",
     "ssb-fixtures": "2.2.0",
     "ssb-keys": "^8.1.0",
     "ssb-ref": "^2.14.3",