[DONT_MERGE] Rocksdb storage

lib/audit.js

@@ -1,23 +1,24 @@
 const hypercoreCrypto = require('hypercore-crypto')
 const flat = require('flat-tree')
-const c = require('compact-encoding')
 const b4a = require('b4a')
 
-const empty = b4a.alloc(32)
+const BitInterlude = require('./bit-interlude')
 
 // this is optimised for speed over mem atm
 // can be tweaked in the future
 
-module.exports = async function auditCore (core) {
+module.exports = async function auditCore (core, storage) {
   const corrections = {
     tree: 0,
     blocks: 0
   }
 
   const length = core.header.tree.length
 
-  const data = await readFullStorage(core.blocks.storage)
-  const tree = await readFullStorage(core.tree.storage)
+  const bitfield = new BitInterlude()
+
+  const data = await readAllBlocks(core.storage)
+  const tree = await readAllTreeNodes(core.tree.storage)
 
   const valid = new Uint8Array(Math.ceil(tree.byteLength / 40))
   const stack = []
@@ -32,8 +33,8 @@ module.exports = async function auditCore (core) {
     if ((node.index & 1) === 0) continue
 
     const [left, right] = flat.children(node.index)
-    const leftNode = getNode(left)
-    const rightNode = getNode(right)
+    const leftNode = tree.get(left)
+    const rightNode = tree.get(right)
 
     if (!rightNode && !leftNode) continue
 
@@ -48,12 +49,8 @@ module.exports = async function auditCore (core) {
       }
     }
 
-    if (leftNode.size) clearNode(leftNode)
-    if (rightNode.size) clearNode(rightNode)
-  }
-
-  if (corrections.tree) {
-    core.tree.cache.clear()
+    if (leftNode.size) clearNode(left)
+    if (rightNode.size) clearNode(right)
   }
 
   let i = 0
@@ -73,57 +70,52 @@ module.exports = async function auditCore (core) {
       try {
         nextOffset = await core.tree.byteOffset(i * 2)
       } catch {
-        core._setBitfield(i, false)
+        storage.deleteBlock(i)
+        bitfield.set(i, false)
         corrections.blocks++
         i++
         continue
       }
     }
 
-    const node = getNode(i * 2)
-    const blk = data.subarray(nextOffset, nextOffset + node.size)
+    const node = tree.get(i * 2)
+    const blk = data.get(i)
     const hash = hypercoreCrypto.data(blk)
 
     nextOffset += blk.byteLength
 
     if (!b4a.equals(hash, node.hash)) {
-      core._setBitfield(i, false)
+      storage.deleteBlock(i)
+      bitfield.set(i, false)
       corrections.blocks++
     }
 
     i++
   }
 
-  return corrections
+  bitfield.flush(storage, core.bitfield)
 
-  function getNode (index) {
-    if (index * 40 + 40 > tree.byteLength) return null
-    const state = { start: index * 40, end: index * 40 + 40, buffer: tree }
-    const size = c.uint64.decode(state)
-    const hash = c.fixed32.decode(state)
-    if (size === 0 && hash.equals(empty)) return null
-    return { index, size, hash }
-  }
+  return corrections
 
   function clearNode (node) {
     valid[node.index] = 0
+    storage.deleteTreeNode(node.index)
+    corrections.tree++
+  }
+}
 
-    if (node.size) {
-      b4a.fill(tree, 0, node.index * 40, node.index * 40 + 40)
-      core.tree.unflushed.set(node.index, core.tree.blankNode(node.index))
-      corrections.tree++
-    }
+async function readAllBlocks (storage) {
+  const data = new Map()
+  for await (const block of storage.createBlockStream()) {
+    data.set(block.index, block.value)
   }
+  return data
 }
 
-function readFullStorage (storage) {
-  return new Promise((resolve, reject) => {
-    storage.stat((_, st) => {
-      if (!st) return resolve(b4a.alloc(0))
-      storage.read(0, st.size, (err, data) => {
-        if (err) reject(err)
-        else resolve(data)
-      })
-    })
-  })
+async function readAllTreeNodes (storage) {
+  const nodes = new Map()
+  for await (const node of storage.createTreeNodeStream()) {
+    nodes.set(node.index, node)
+  }
+  return nodes
 }