autobase - npm Package Compare versions

Comparing version 6.1.7 to 6.2.0

index.js

@@ -63,2 +63,3 @@ const b4a = require('b4a')
     this.localWriter = null
+    this.isIndexer = false
 
@@ -147,4 +148,6 @@ this.activeWriters = new ActiveWriters()
 
+    this._initialHeads = []
     this._initialSystem = null
     this._initialViews = null
+
     this._waiting = new SignalPromise()
@@ -184,4 +187,4 @@
 
-  get isIndexer () {
-    return this.localWriter ? this.localWriter.isIndexer : false
+  get isActiveIndexer () {
+    return this.localWriter ? this.localWriter.isActiveIndexer : false
   }
@@ -277,3 +280,3 @@
 
-    const { bootstrap, system } = await this._loadSystemInfo()
+    const { bootstrap, system, heads } = await this._loadSystemInfo()
 
@@ -287,3 +290,5 @@ this.version = system
     this.bootstrap = bootstrap
+
     this._initialSystem = system
+    this._initialHeads = heads
 
@@ -296,5 +301,5 @@ await this._makeLinearizer(system)
     const bootstrap = this.bootstrap || (await this.local.getUserData('referrer')) || this.local.key
-    if (!pointer) return { bootstrap, system: null }
+    if (!pointer) return { bootstrap, system: null, heads: [] }
 
-    const { indexed, views } = c.decode(messages.BootRecord, pointer)
+    const { indexed, views, heads } = c.decode(messages.BootRecord, pointer)
     const { key, length } = indexed
@@ -304,3 +309,3 @@
 
-    if (!length) return { bootstrap, system: null }
+    if (!length) return { bootstrap, system: null, heads: [] }
 
@@ -318,3 +323,3 @@ const encryptionKey = AutoStore.getBlockKey(bootstrap, this.encryptionKey, '_system')
       this._systemPointer = 0
-      return { bootstrap, system: null }
+      return { bootstrap, system: null, heads: [] }
     }
@@ -337,3 +342,4 @@
       bootstrap,
-      system
+      system,
+      heads
     }
@@ -481,4 +487,62 @@ }
     this.queueFastForward()
+
+    await this._catchup(this._initialHeads)
   }
 
+  async _catchup (nodes) {
+    if (!nodes.length) return
+
+    const visited = new Set()
+    const writers = new Map()
+
+    while (nodes.length) {
+      const { key, length } = nodes.pop()
+
+      const hex = b4a.toString(key, 'hex')
+      const ref = hex + ':' + length
+
+      if (visited.has(ref)) continue
+      visited.add(ref)
+
+      let w = writers.get(hex)
+      if (!w) {
+        const writer = await this._getWriterByKey(key, -1, 0, true, false, null)
+
+        w = { writer, end: writer.length }
+
+        writers.set(hex, w)
+      }
+
+      if (w.writer.length >= length) continue
+
+      if (length > w.end) w.end = length
+
+      const block = await w.writer.core.get(length - 1)
+
+      for (const dep of block.node.heads) {
+        nodes.push(dep)
+      }
+    }
+
+    while (writers.size) {
+      for (const [hex, info] of writers) {
+        const { writer, end } = info
+
+        if (writer === null || writer.length === end) {
+          writers.delete(hex)
+          continue
+        }
+
+        if (writer.available <= writer.length) await writer.update()
+
+        const node = writer.advance()
+        if (!node) continue
+
+        this.linearizer.addHead(node)
+      }
+    }
+
+    await this._drain() // runs for one tick
+  }
+
   _reindexersIdle () {
@@ -531,3 +595,3 @@ for (const idx of this.linearizer.indexers) {
     if (this._primaryBootstrap) await this._primaryBootstrap.close()
-    for (const w of this.activeWriters) await w.close()
+    await this.activeWriters.clear()
     await this.corePool.clear()
@@ -569,3 +633,3 @@ await this.store.close()
 
-      if (!unqueued || w.isIndexer || this.localWriter === w) continue
+      if (!unqueued || w.isActiveIndexer || this.localWriter === w) continue
 
@@ -663,3 +727,3 @@ await this._closeWriter(w, false)
 
-    const isIndexer = this.localWriter.isIndexer || isPendingIndexer
+    const isIndexer = this.localWriter.isActiveIndexer || isPendingIndexer
 
@@ -701,4 +765,2 @@ if (!isIndexer || this._acking || this.closing) return
 
-    await this._advanced // ensure all local state has been applied, only needed until persistent batches
-
     // if a reset is scheduled await those
@@ -711,8 +773,2 @@ while (this._queueViewReset && !this.closing) await this._bump()
 
-    // make sure all local nodes are processed before continuing
-    while (!this.closing && this.localWriter.core.length > this.localWriter.length) {
-      await this.localWriter.waitForSynced()
-      await this._bump() // make sure its all flushed...
-    }
-
     if (this._appending === null) this._appending = []
@@ -726,3 +782,7 @@
 
-    await this._bump()
+    // await in case append is in current tick
+    if (this._advancing) await this._advancing
+
+    // only bump if there are unflushed nodes
+    if (this._appending !== null) return this._bump()
   }
@@ -919,3 +979,3 @@
     this.linearizer = new Linearizer(indexers, { heads, writers: this.activeWriters })
-    this._addCheckpoints = !!(this.localWriter && (this.localWriter.isIndexer || this._isPending()))
+    this._addCheckpoints = !!(this.localWriter && (this.localWriter.isActiveIndexer || this._isPending()))
     this._updateAckThreshold()
@@ -939,3 +999,4 @@ }
     this._checkWriters.push(bootstrap)
-    bootstrap.isIndexer = true
+    if (bootstrap === this.localWriter) this._setLocalIndexer()
+    bootstrap.isActiveIndexer = true
     bootstrap.inflateBackground()
@@ -963,3 +1024,3 @@ await bootstrap.ready()
       const writer = await this._getWriterByKey(head.key, head.length, 0, false, false, sys)
-      writer.isIndexer = true
+      writer.isActiveIndexer = true
       writer.inflateBackground()
@@ -998,3 +1059,3 @@ indexers.push(writer)
 
-    if (!this.localWriter || !this.localWriter.isIndexer) return
+    if (!this.localWriter || !this.localWriter.isActiveIndexer) return
 
@@ -1021,3 +1082,3 @@ if (!hasWriter(this.linearizer.indexers, this.localWriter)) {
     this._closeWriter(this.localWriter, true)
-    if (this.localWriter.isIndexer) this._clearLocalIndexer()
+    if (this.localWriter.isActiveIndexer) this._clearLocalIndexer()
 
@@ -1030,6 +1091,19 @@ this.localWriter = null
 
+  _setLocalIndexer () {
+    assert(this.localWriter !== null)
+    this.isIndexer = true
+    this._addCheckpoints = true // unset once indexer is cleared
+    this.emit('is-indexer')
+  }
+
+  _unsetLocalIndexer () {
+    assert(this.localWriter !== null)
+    this.isIndexer = false
+    this.emit('is-non-indexer')
+  }
+
   _clearLocalIndexer () {
     if (!this.localWriter) return
 
-    this.localWriter.isIndexer = false
+    this.localWriter.isActiveIndexer = false
 
@@ -1104,8 +1178,3 @@ if (this._ackTimer) this._ackTimer.stop()
 
-    const cores = this._viewStore.getIndexedCores()
-    const views = new Array(cores.length - 1)
-    for (const core of cores) {
-      if (core.systemIndex === -1) continue
-      views[core.systemIndex] = core.name
-    }
+    const views = this._viewStore.indexedViewsByName()
 
@@ -1115,2 +1184,10 @@ await this._setBootRecord(this.system.core.key, length, this.system.heads, views)
 
+  async _updateBootRecordHeads (heads) {
+    if (this._systemPointer === 0) return // first tick
+
+    const views = this._viewStore.indexedViewsByName()
+
+    await this._setBootRecord(this.system.core.key, this._systemPointer, heads, views)
+  }
+
   async _setBootRecord (key, length, heads, views) {
@@ -1128,3 +1205,3 @@ const pointer = c.encode(messages.BootRecord, {
     while (!this.closing) {
-      if (this.fastForwardTo !== null) {
+      if (this.opened && this.fastForwardTo !== null) {
         await this._applyFastForward()
@@ -1139,4 +1216,4 @@ this.system.requestWakeup()
 
-      const remoteAdded = await this._addRemoteHeads()
-      const localNodes = this._appending === null ? null : this._addLocalHeads()
+      const remoteAdded = this.opened ? await this._addRemoteHeads() : null
+      const localNodes = this.opened && this._appending !== null ? this._addLocalHeads() : null
 
@@ -1156,2 +1233,4 @@ if (this._maybeStaticFastForward === true && this.fastForwardEnabled === true) await this._checkStaticFastForward()
 
+      if (this.opened) await this._updateBootRecordHeads(this.system.heads)
+
       if (this.localWriter !== null && localNodes !== null) {
@@ -1161,3 +1240,3 @@ await this._flushLocal(localNodes)
 
-      if (this._pendingLocalRemoval && !this.localWriter.isIndexer) this._unsetLocalWriter()
+      if (this._pendingLocalRemoval && !this.localWriter.isActiveIndexer) this._unsetLocalWriter()
 
@@ -1184,2 +1263,3 @@ if (this.closing) return
           await this._gcWriters()
+          if (!this.opened) break // at most one tick preready
           continue // rerun the update loop as a writer might have been added
@@ -1276,3 +1356,3 @@ }
 
-    await this._gcWriters()
+    if (!this.closing) await this._gcWriters()
   }
@@ -1766,3 +1846,6 @@
     // If we are getting added as indexer, already start adding checkpoints while we get confirmed...
-    if (writer === this.localWriter && isIndexer) this._addCheckpoints = true
+    if (writer === this.localWriter) {
+      if (isIndexer) this._setLocalIndexer()
+      else this._unsetLocalIndexer() // unset if demoted
+    }
 
@@ -1778,3 +1861,6 @@ // fetch any nodes needed for dependents
 
-    if (b4a.equals(key, this.local.key)) this._pendingLocalRemoval = true
+    if (b4a.equals(key, this.local.key)) {
+      this._pendingLocalRemoval = true
+      if (this.isIndexer) this._unsetLocalIndexer()
+    }
 
@@ -1781,0 +1867,0 @@ const w = this.activeWriters.get(key)

lib/active-writers.js

@@ -31,2 +31,10 @@ const b4a = require('b4a')
   }
+
+  clear () {
+    const p = []
+    for (const w of this.map.values()) p.push(w.close())
+    this.map.clear()
+
+    return Promise.all(p)
+  }
 }

lib/consensus.js

@@ -27,3 +27,3 @@ const BufferMap = require('tiny-buffer-map')
   addHead (node) {
-    if (!node.writer.isIndexer) return
+    if (!node.writer.isActiveIndexer) return
     if (this._isMerge(node)) this.merges.add(node)
@@ -56,3 +56,3 @@ this.updated = true
   _isMerge (node) {
-    if (!node.writer.isIndexer) return false
+    if (!node.writer.isActiveIndexer) return false
 
@@ -156,3 +156,3 @@ const deps = []
   _acks (target) {
-    const acks = target.writer.isIndexer ? [target] : [] // TODO: can be cached on the target node in future (ie if we add one we dont have to check it again)
+    const acks = target.writer.isActiveIndexer ? [target] : [] // TODO: can be cached on the target node in future (ie if we add one we dont have to check it again)
 
@@ -159,0 +159,0 @@ for (const idx of this.indexers) {

lib/extension.js

@@ -52,3 +52,3 @@ const c = require('compact-encoding')
     for (const w of this.base.activeWriters) {
-      if (w.isIndexer || w.flushed()) continue
+      if (w.isActiveIndexer || w.flushed()) continue
       writers.push({ key: w.core.key, length: w.length })
@@ -55,0 +55,0 @@ }

lib/linearizer.js

@@ -53,3 +53,3 @@ const b4a = require('b4a')
 
-    if (this.writer.isIndexer) this.clock.set(this.writer.core.key, this.length)
+    if (this.writer.isActiveIndexer) this.clock.set(this.writer.core.key, this.length)
   }
@@ -125,3 +125,3 @@
     this.tip.add(node)
-    if (node.writer.isIndexer) this.consensus.addHead(node)
+    if (node.writer.isActiveIndexer) this.consensus.addHead(node)
 
@@ -165,3 +165,3 @@ this.size++
   shouldAck (writer, pending = false) {
-    if (!writer || !writer.isIndexer) return false
+    if (!writer || !writer.isActiveIndexer) return false
 
@@ -177,3 +177,3 @@ // all indexers have to flushed to the dag before we ack as a quick "debounce"
     for (const head of this.heads) {
-      if (!head.writer.isIndexer) return true
+      if (!head.writer.isActiveIndexer) return true
       if (head.writer === writer) isHead = true
@@ -180,0 +180,0 @@ }

lib/store.js

@@ -130,2 +130,14 @@ const b4a = require('b4a')
 
+  indexedViewsByName () {
+    const views = []
+
+    for (let i = 0; i < this.base.system.views.length; i++) {
+      const core = this.getByIndex(i)
+      if (!core || !core.pendingIndexedLength) break
+      views.push(core.name)
+    }
+
+    return views
+  }
+
   async flush () {
@@ -132,0 +144,0 @@ while (this.waiting.length) {

lib/writer.js

@@ -24,2 +24,3 @@ const Linearizer = require('./linearizer')
     this.isIndexer = false
+    this.isActiveIndexer = false
     this.available = length
@@ -88,3 +89,3 @@ this.length = length
 
-    if (this.core.length === 0 || !this.isIndexer || this.nodes.length === 0) {
+    if (this.core.length === 0 || !this.isActiveIndexer || this.nodes.length === 0) {
       return
@@ -206,3 +207,3 @@ }
 
-    if (this.digestLength < this.core.length && this.isIndexer) await this._checkDigest()
+    if (this.digestLength < this.core.length && this.isActiveIndexer) await this._checkDigest()
 
@@ -347,3 +348,3 @@ return this.length < this.available
 
-    if (this.isIndexer && checkpoint) {
+    if (this.isActiveIndexer && checkpoint) {
       this._addCheckpoints(checkpoint)
@@ -350,0 +351,0 @@ }

package.json

 {
   "name": "autobase",
-  "version": "6.1.7",
+  "version": "6.2.0",
   "description": "",
@@ -5,0 +5,0 @@ "main": "index.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

184866

increased by1.41%

Lines of code

4987

increased by1.55%

No dependency changes