eagerly flush hints (#745)

* eagerly flush hints

* fix test and only for default

* debounce

* lint

Author: mafintosh

lib/core.js

@@ -55,6 +55,8 @@ module.exports = class Core {
     this.destroyed = false
     this.closed = false
 
+    this.hintsChanged = false
+
     this._bitfield = null
     this._verifies = null
     this._verifiesFlushed = null
@@ -531,6 +533,8 @@ module.exports = class Core {
           this.replicator.onhave(bitfield.start, bitfield.length, bitfield.drop)
         }
       }
+
+      if (this.hintsChanged) await this.flushHints()
     } finally {
       this.state._clearActiveBatch()
       this.state.mutex.unlock()
@@ -539,6 +543,20 @@ module.exports = class Core {
     return verifies[0] !== null
   }
 
+  async flushHints() {
+    if (!this.hintsChanged) return
+    this.hintsChanged = false // we unset this immediately as a "debounce"
+
+    const tx = this.state.storage.write()
+
+    tx.setHints({
+      contiguousLength: this.header.hints.contiguousLength,
+      remoteContiguousLength: this.header.hints.remoteContiguousLength
+    })
+
+    await tx.flush()
+  }
+
   async checkConflict(proof, from) {
     if (this.state.length < proof.upgrade.length || proof.fork !== this.state.fork) {
       // out of date this proof - ignore for now
@@ -716,13 +734,13 @@ module.exports = class Core {
 
     if (contig.length !== -1 && contig.length !== this.header.hints.contiguousLength) {
       this.header.hints.contiguousLength = contig.length
+      this.hintsChanged = true
     }
   }
 
-  async updateRemoteContiguousLength(length) {
+  updateRemoteContiguousLength(length) {
     this.header.hints.remoteContiguousLength = length
-    await this.state.flushHints()
-    if (this.header.hints.remoteContiguousLength !== length) return
+    this.hintsChanged = true
 
     for (let i = this.monitors.length - 1; i >= 0; i--) {
       this.monitors[i].emit('remote-contiguous-length', length)

lib/replicator.js

@@ -1166,7 +1166,7 @@ class Peer {
           this.remoteFork === this.core.state.fork &&
           length > this.core.header.hints.remoteContiguousLength
         ) {
-          await this.core.updateRemoteContiguousLength(length)
+          this.core.updateRemoteContiguousLength(length)
         }
       }
     } else if (length === 1) {
@@ -1184,6 +1184,7 @@ class Peer {
       }
     }
 
+    if (this.core.hintsChanged) await this.core.flushHints()
     if (drop === false) this._update()
   }
 

lib/session-state.js

@@ -300,6 +300,8 @@ module.exports = class SessionState {
       const append = b ? { start: b.start, length: b.appends, drop: false } : null
 
       this.onappend(tree, append, true)
+
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
     } finally {
       this.mutex.unlock()
       this.core.checkIfIdle()
@@ -360,6 +362,8 @@ module.exports = class SessionState {
 
         this.onappend(head, bitfield, flushed)
       }
+
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
     } finally {
       this._clearActiveBatch()
       this.updating = false
@@ -419,6 +423,8 @@ module.exports = class SessionState {
       if (dependency) this.storage.setDependencyHead(dependency)
 
       this.ontruncate(tree, tree.length, batch.treeLength, flushed)
+
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
     } finally {
       this._unlock()
     }
@@ -445,6 +451,8 @@ module.exports = class SessionState {
       if (dependency) this.storage.setDependencyHead(dependency)
 
       this.ontruncate(tree, batch.ancestors, batch.treeLength, flushed)
+
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
     } finally {
       this._unlock()
     }
@@ -487,23 +495,6 @@ module.exports = class SessionState {
     return { dependency, tree, roots: batch.roots }
   }
 
-  async flushHints() {
-    await this.mutex.lock()
-
-    try {
-      const tx = this.createWriteBatch()
-
-      tx.setHints({
-        contiguousLength: this.core.header.hints.contiguousLength,
-        remoteContiguousLength: this.core.header.hints.remoteContiguousLength
-      })
-
-      await tx.flush()
-    } finally {
-      this._unlock()
-    }
-  }
-
   async clear(start, end, cleared) {
     await this.mutex.lock()
 
@@ -548,9 +539,12 @@ module.exports = class SessionState {
       // todo: atomic event handle
       if (this.isDefault() && flushed) {
         const length = end - start
+
         this.core.updateContiguousLength({ start, length, drop: true })
         this.core._setBitfieldRanges(start, end, false)
         this.core.replicator.onhave(start, length, true)
+
+        if (this.core.hintsChanged) await this.core.flushHints()
       }
     } finally {
       this._unlock()
@@ -631,6 +625,8 @@ module.exports = class SessionState {
 
       this.onappend(tree, bitfield, flushed)
 
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
+
       return { length: this.length, byteLength: this.byteLength }
     } finally {
       this._unlock()
@@ -780,6 +776,8 @@ module.exports = class SessionState {
 
       if (truncating) this.ontruncate(tree, sharedLength, origLength, flushed)
       if (sharedLength < length) this.onappend(tree, null, flushed)
+
+      if (this.core.hintsChanged && this.isDefault()) await this.core.flushHints()
     } finally {
       this.mutex.unlock()
     }

test/replicate.js

@@ -1773,6 +1773,10 @@ test('session id reuse does not stall', async function (t) {
     if (all.length === 0) break
   }
 
+  // wait a little bit, cause technically the above has a storage race
+  // since it checks the bitfield manually and the event isnt timed to that
+  await new Promise((resolve) => setTimeout(resolve, 1000))
+
   t.pass('All blocks downloaded')
   t.is(downloaded, 100, 'Downloaded all blocks exactly once')