fix reentry on contig update (#119)

* fix reentry on contig update

* fix non-sparse truncations

* move events further down for safety

Author: mafintosh

index.js

@@ -312,8 +312,7 @@ module.exports = class Hypercore extends EventEmitter {
       crypto: this.crypto,
       legacy: opts.legacy,
       auth: opts.auth,
-      onupdate: this._oncoreupdate.bind(this),
-      oncontigupdate: this._oncorecontigupdate.bind(this)
+      onupdate: this._oncoreupdate.bind(this)
     })
 
     if (opts.userData) {
@@ -496,30 +495,38 @@ module.exports = class Hypercore extends EventEmitter {
 
   _oncoreupdate (status, bitfield, value, from) {
     if (status !== 0) {
-      const truncated = (status & 0b10) !== 0
-      const appended = (status & 0b01) !== 0
+      const truncatedNonSparse = (status & 0b1000) !== 0
+      const appendedNonSparse = (status & 0b0100) !== 0
+      const truncated = (status & 0b0010) !== 0
+      const appended = (status & 0b0001) !== 0
 
       if (truncated) {
         this.replicator.ontruncate(bitfield.start)
       }
 
+      if ((status & 0b0011) !== 0) {
+        this.replicator.onupgrade()
+      }
+
       for (let i = 0; i < this.sessions.length; i++) {
         const s = this.sessions[i]
 
         if (truncated) {
           if (s.cache) s.cache.clear()
-          s.emit('truncate', bitfield.start, this.core.tree.fork)
+
           // If snapshotted, make sure to update our compat so we can fail gets
           if (s._snapshot && bitfield.start < s._snapshot.compatLength) s._snapshot.compatLength = bitfield.start
         }
 
-        // For sparse sessions, immediately emit appends. Non-sparse sessions
-        // are handled separately and only emit appends when their contiguous
-        // length is updated.
-        if (appended && s.sparse) s.emit('append')
-      }
+        if (s.sparse ? truncated : truncatedNonSparse) {
+          s.emit('truncate', bitfield.start, this.core.tree.fork)
+        }
 
-      this.replicator.onupgrade()
+        // For sparse sessions, immediately emit appends. If non-sparse, emit if contig length has updated
+        if (s.sparse ? appended : appendedNonSparse) {
+          s.emit('append')
+        }
+      }
     }
 
     if (bitfield) {
@@ -535,16 +542,6 @@ module.exports = class Hypercore extends EventEmitter {
     }
   }
 
-  _oncorecontigupdate () {
-    // For non-sparse sessions, emit appends only when the contiguous length is
-    // updated.
-    for (let i = 0; i < this.sessions.length; i++) {
-      const s = this.sessions[i]
-
-      if (!s.sparse) s.emit('append')
-    }
-  }
-
   _onpeerupdate (added, peer) {
     const name = added ? 'peer-add' : 'peer-remove'
 

lib/core.js

@@ -9,9 +9,8 @@ const { BAD_ARGUMENT, STORAGE_EMPTY, STORAGE_CONFLICT, INVALID_SIGNATURE } = req
 const m = require('./messages')
 
 module.exports = class Core {
-  constructor (header, crypto, oplog, tree, blocks, bitfield, auth, legacy, onupdate, oncontigupdate) {
+  constructor (header, crypto, oplog, tree, blocks, bitfield, auth, legacy, onupdate) {
     this.onupdate = onupdate
-    this.oncontigupdate = oncontigupdate
     this.header = header
     this.crypto = crypto
     this.oplog = oplog
@@ -27,8 +26,6 @@ module.exports = class Core {
     this._verifiesFlushed = null
     this._mutex = new Mutex()
     this._legacy = legacy
-
-    this._updateContiguousLength(header.contiguousLength)
   }
 
   static async open (storage, opts = {}) {
@@ -134,6 +131,11 @@ module.exports = class Core {
       await bitfield.clear()
     }
 
+    // compat from earlier version that do not store contig length
+    if (header.contiguousLength === 0) {
+      while (bitfield.get(header.contiguousLength)) header.contiguousLength++
+    }
+
     const auth = opts.auth || this.createAuth(crypto, header.signer)
 
     for (const e of entries) {
@@ -149,6 +151,7 @@ module.exports = class Core {
 
       if (e.bitfield) {
         bitfield.setRange(e.bitfield.start, e.bitfield.length, !e.bitfield.drop)
+        updateContig(header, e.bitfield, bitfield)
       }
 
       if (e.treeUpgrade) {
@@ -165,7 +168,7 @@ module.exports = class Core {
       }
     }
 
-    return new this(header, crypto, oplog, tree, blocks, bitfield, auth, !!opts.legacy, opts.onupdate || noop, opts.oncontigupdate || noop)
+    return new this(header, crypto, oplog, tree, blocks, bitfield, auth, !!opts.legacy, opts.onupdate || noop)
   }
 
   _shouldFlush () {
@@ -196,17 +199,6 @@ module.exports = class Core {
     await this.blocks.put(index, value, byteOffset)
   }
 
-  _updateContiguousLength (index, length = 0) {
-    if (index === this.header.contiguousLength) {
-      let i = this.header.contiguousLength + length
-
-      while (this.bitfield.get(i)) i++
-
-      this.header.contiguousLength = i
-      this.oncontigupdate()
-    }
-  }
-
   async userData (key, value) {
     // TODO: each oplog append can set user data, so we should have a way
     // to just hitch a ride on one of the other ongoing appends?
@@ -286,12 +278,12 @@ module.exports = class Core {
       this.bitfield.setRange(batch.ancestors, batch.length - batch.ancestors, true)
       batch.commit()
 
-      this.header.contiguousLength = batch.length
-      this.oncontigupdate()
       this.header.tree.length = batch.length
       this.header.tree.rootHash = hash
       this.header.tree.signature = batch.signature
-      this.onupdate(0b01, entry.bitfield, null, null)
+
+      const status = 0b0001 | updateContig(this.header, entry.bitfield, this.bitfield)
+      this.onupdate(status, entry.bitfield, null, null)
 
       if (this._shouldFlush()) await this._flushOplog()
 
@@ -329,9 +321,11 @@ module.exports = class Core {
 
       await this.oplog.append([entry], false)
 
+      let status = 0b0001
+
       if (bitfield) {
         this.bitfield.set(bitfield.start, true)
-        this._updateContiguousLength(bitfield.start, bitfield.length)
+        status |= updateContig(this.header, bitfield, this.bitfield)
       }
 
       batch.commit()
@@ -340,7 +334,8 @@ module.exports = class Core {
       this.header.tree.length = batch.length
       this.header.tree.rootHash = batch.rootHash
       this.header.tree.signature = batch.signature
-      this.onupdate(0b01, bitfield, value, from)
+
+      this.onupdate(status, bitfield, value, from)
 
       if (this._shouldFlush()) await this._flushOplog()
     } finally {
@@ -387,14 +382,16 @@ module.exports = class Core {
           continue
         }
 
+        let status = 0
+
         if (bitfield) {
           this.bitfield.set(bitfield.start, true)
-          this._updateContiguousLength(bitfield.start, bitfield.length)
+          status = updateContig(this.header, bitfield, this.bitfield)
         }
 
         batch.commit()
 
-        this.onupdate(0, bitfield, value, from)
+        this.onupdate(status, bitfield, value, from)
       }
 
       if (this._shouldFlush()) await this._flushOplog()
@@ -472,15 +469,15 @@ module.exports = class Core {
     addReorgHint(this.header.hints.reorgs, this.tree, batch)
     batch.commit()
 
-    const appended = batch.length > batch.ancestors
+    const contigStatus = updateContig(this.header, entry.bitfield, this.bitfield)
+    const status = ((batch.length > batch.ancestors) ? 0b0011 : 0b0010) | contigStatus
 
-    this.header.contiguousLength = Math.min(batch.ancestors, this.header.contiguousLength)
-    this.oncontigupdate()
     this.header.tree.fork = batch.fork
     this.header.tree.length = batch.length
     this.header.tree.rootHash = batch.hash()
     this.header.tree.signature = batch.signature
-    this.onupdate(appended ? 0b11 : 0b10, entry.bitfield, null, from)
+
+    this.onupdate(status, entry.bitfield, null, from)
 
     // TODO: there is a bug in the merkle tree atm where it cannot handle unflushed
     // truncates if we append or download anything after the truncation point later on
@@ -501,6 +498,36 @@ module.exports = class Core {
   }
 }
 
+function updateContig (header, upd, bitfield) {
+  const end = upd.start + upd.length
+
+  let c = header.contiguousLength
+
+  if (upd.drop) {
+    // If we dropped a block in the current contig range, "downgrade" it
+    if (c <= end && c > upd.start) {
+      c = upd.start
+    }
+  } else {
+    if (c <= end && c >= upd.start) {
+      c = end
+      while (bitfield.get(c)) c++
+    }
+  }
+
+  if (c === header.contiguousLength) {
+    return 0b0000
+  }
+
+  if (c > header.contiguousLength) {
+    header.contiguousLength = c
+    return 0b0100
+  }
+
+  header.contiguousLength = c
+  return 0b1000
+}
+
 function addReorgHint (list, tree, batch) {
   if (tree.length === 0 || tree.fork === batch.fork) return
 

test/core.js

@@ -178,7 +178,7 @@ test('core - update hook is triggered', async function (t) {
   let ran = 0
 
   core.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b01, 'was appended')
+    t.ok(status & 0b01, 'was appended')
     t.is(from, null, 'was local')
     t.alike(bitfield, { drop: false, start: 0, length: 4 })
     ran |= 1
@@ -189,7 +189,7 @@ test('core - update hook is triggered', async function (t) {
   const peer = {}
 
   clone.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b01, 'was appended')
+    t.ok(status & 0b01, 'was appended')
     t.is(from, peer, 'was remote')
     t.alike(bitfield, { drop: false, start: 1, length: 1 })
     t.alike(value, Buffer.from('b'))
@@ -217,7 +217,7 @@ test('core - update hook is triggered', async function (t) {
   }
 
   core.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b10, 'was truncated')
+    t.ok(status & 0b10, 'was truncated')
     t.is(from, null, 'was local')
     t.alike(bitfield, { drop: true, start: 1, length: 3 })
     ran |= 8
@@ -226,7 +226,7 @@ test('core - update hook is triggered', async function (t) {
   await core.truncate(1, 1)
 
   core.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b01, 'was appended')
+    t.ok(status & 0b01, 'was appended')
     t.is(from, null, 'was local')
     t.alike(bitfield, { drop: false, start: 1, length: 1 })
     ran |= 16
@@ -235,7 +235,7 @@ test('core - update hook is triggered', async function (t) {
   await core.append([Buffer.from('e')])
 
   clone.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b11, 'was appended and truncated')
+    t.ok(status & 0b11, 'was appended and truncated')
     t.is(from, peer, 'was remote')
     t.alike(bitfield, { drop: true, start: 1, length: 3 })
     ran |= 32
@@ -248,7 +248,7 @@ test('core - update hook is triggered', async function (t) {
   }
 
   core.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b10, 'was truncated')
+    t.ok(status & 0b10, 'was truncated')
     t.is(from, null, 'was local')
     t.alike(bitfield, { drop: true, start: 1, length: 1 })
     ran |= 64
@@ -257,7 +257,7 @@ test('core - update hook is triggered', async function (t) {
   await core.truncate(1, 2)
 
   clone.onupdate = (status, bitfield, value, from) => {
-    t.is(status, 0b10, 'was truncated')
+    t.ok(status & 0b10, 'was truncated')
     t.is(from, peer, 'was remote')
     t.alike(bitfield, { drop: true, start: 1, length: 1 })
     ran |= 128