Add core.remoteContiguousLength & remote-contiguous-length event (#743)

* wip

* Apply prettier to remoteContiguousLength commits thus far

* Add event listener to `remote contiguous length` test

* Add test to showcase remote contig length only updates for fully contig

To save messages being sent by all peers, the existing behavior of
sending a `range` request in `broadcastRange()` only when fully
contiguous is being kept. This means the `.remoteContiguousLength` and
the `remote-contiguous-length` event only update / fire when the remote
peer is fully contiguous and not when partially contiguous.

* Rename `contig` var to `fullyContig`

This emphasizes that the variable is true only when it is fully
contiguous, ie having the full range of blocks. This is in contrast to
being partially contiguous.

* Add test to verify truncation updates `remoteContiguousLength` too

* Update tests to include `length` arg in `remote-contiguous-length` event

* Document `.remoteContiguousLength` in README.md

Includes note that it is only updated when the peer thinks it is fully
contiguous (aka non-sparse).

* Document `remote-contiguous-length` event

* Clarify that truncating doesnt update `.remoteContiguousLength` anywhere

* Test that appends after truncating dont fire remote contig length event

* Format remote contig length event after truncate test

* Fix remoteContiguousLength hint to update on truncating

Now allows appends after truncating to fire remote contig event.

* Add test for persisting remoteContiguousLength between reloads

* Parse remoteContiguousLength when parsing header from storage

Fixes the `remote contiguous length - persists` test assuming storage
support.

* Use `truncate` & `download` events instead of timeouts in test

* Ensure contiguous lengths default to 0 when parsing header

Not likely to happen in practice, but just to be safe.

* Update remoteContiguousLength only on range starting at `0`

Peers always send `range` requests starting with `0` if they are
contiguous. Partial ranges would imply they are not contiguous so
shouldn't be able to sparsely update the `remoteContiguousLength`.

Because there is no need check the `start` of the range &
`core.updateRemoteContiguousLength()` already checks if the `length` is
larger, the logic is simplified to only call
`updateRemoteContiguousLength()` when updating the
`_remoteContiguousLength` for the peer when the range isnt a drop.

* Use `done()` to await downloads in remote contig tests

* Remove `.catch(noop)` of `.updateRemoteContiguousLength()`

Could mask underlying error and its assumed that storage errors at this
point should not be caught.

* Add test to verify remote contig length when peer truncates & appends

Should test if a reorg will correctly update the remote contiguous
length to match.

* Skip updating `remoteContiguousLength` when remote is on older fork

* Try/catch & pause if updating remote contig hint throws

* Move `updateRemoteContiguousLength()` next to `updateContiguousLength()`

* Remove try catch to let error bubble up from updating remote contig

* Move remote length check into `onrange` to avoid promise allocation

`core.updateRemoteContiguousLength()` is only called from `onrange` so
moving the check doesn't allow unviable lengths through from elsewhere.

* Change `info.hints` check style when parsing header

Use ternary instead of a combination of optional chaining and null
coalescing.

* Fix formatting

* must be same fork to update contig

---------

Co-authored-by: Mathias Buus <[email protected]>

Author: lejeunerenard

README.md

@@ -594,7 +594,13 @@ Populated after `ready` has been emitted. Will be `0` before the event.
 
 #### `core.contiguousLength`
 
-How many blocks are contiguously available starting from the first block of this core?
+How many blocks are contiguously available starting from the first block of this core.
+
+Populated after `ready` has been emitted. Will be `0` before the event.
+
+#### `core.remoteContiguousLength`
+
+How many blocks are contiguously available starting from the first block of this core on any known remote. This is only updated when a remote thinks it is fully contiguous such that they have all known blocks.
 
 Populated after `ready` has been emitted. Will be `0` before the event.
 
@@ -708,6 +714,10 @@ Emitted when a block is uploaded to a peer.
 
 Emitted when a block is downloaded from a peer.
 
+#### `core.on('remote-contiguous-length', length)`
+
+Emitted when the max known contiguous `length` from a remote, ie `core.remoteContiguousLength`, is updated. Note this is not emitted when core is truncated.
+
 #### `Hypercore.MAX_SUGGESTED_BLOCK_SIZE`
 
 The constant for max size (15MB) for blocks appended to Hypercore. This max ensures blocks are replicated smoothly.

index.js

@@ -564,6 +564,11 @@ class Hypercore extends EventEmitter {
     return this.state.byteLength - this.state.length * this.padding
   }
 
+  get remoteContiguousLength() {
+    if (this.opened === false) return 0
+    return Math.min(this.core.state.length, this.core.header.hints.remoteContiguousLength)
+  }
+
   get contiguousLength() {
     if (this.opened === false) return 0
     return Math.min(this.core.state.length, this.core.header.hints.contiguousLength)

lib/core.js

@@ -200,7 +200,8 @@ module.exports = class Core {
         },
         hints: {
           reorgs: [],
-          contiguousLength: 0
+          contiguousLength: 0,
+          remoteContiguousLength: 0
         }
       }
 
@@ -284,7 +285,7 @@ module.exports = class Core {
     if (header.hints.contiguousLength !== len) {
       header.hints.contiguousLength = len
       const tx = storage.write()
-      tx.setHints({ contiguousLength: len })
+      tx.setHints(header.hints)
       await tx.flush()
     }
 
@@ -718,6 +719,16 @@ module.exports = class Core {
     }
   }
 
+  async updateRemoteContiguousLength(length) {
+    this.header.hints.remoteContiguousLength = length
+    await this.state.flushHints()
+    if (this.header.hints.remoteContiguousLength !== length) return
+
+    for (let i = this.monitors.length - 1; i >= 0; i--) {
+      this.monitors[i].emit('remote-contiguous-length', length)
+    }
+  }
+
   onappend(tree, bitfield) {
     this.header.tree = tree
 
@@ -860,7 +871,8 @@ function parseHeader(info) {
     tree: info.head || getDefaultTree(),
     hints: {
       reorgs: [],
-      contiguousLength: info.hints ? info.hints.contiguousLength : 0
+      contiguousLength: info.hints ? info.hints.contiguousLength : 0,
+      remoteContiguousLength: info.hints ? info.hints.remoteContiguousLength : 0
     }
   }
 }

lib/replicator.js

@@ -514,19 +514,19 @@ class Peer {
     else this._clearLocalRange(start, length)
 
     const i = Math.floor(start / DEFAULT_SEGMENT_SIZE)
-    const contig = this.core.header.hints.contiguousLength === this.core.state.length
+    const fullyContig = this.core.header.hints.contiguousLength === this.core.state.length
 
     if (
       start + LAST_BLOCKS < this.core.state.length &&
       !this.remoteSegmentsWanted.has(i) &&
       !drop &&
-      !contig
+      !fullyContig
     ) {
       return
     }
 
     let force = false
-    if (contig && !drop) {
+    if (fullyContig && !drop) {
       start = 0
       length = this.core.state.length
 
@@ -1152,15 +1152,23 @@ class Peer {
     this._clearLocalRange(fixedStart, length)
   }
 
-  onrange({ drop, start, length }) {
+  async onrange({ drop, start, length }) {
     const has = drop === false
 
     if (drop === true && start < this._remoteContiguousLength) {
       this._remoteContiguousLength = start
     }
 
     if (start === 0 && drop === false) {
-      if (length > this._remoteContiguousLength) this._remoteContiguousLength = length
+      if (length > this._remoteContiguousLength) {
+        this._remoteContiguousLength = length
+        if (
+          this.remoteFork === this.core.state.fork &&
+          length > this.core.header.hints.remoteContiguousLength
+        ) {
+          await this.core.updateRemoteContiguousLength(length)
+        }
+      }
     } else if (length === 1) {
       const bitfield = this.core.skipBitfield === null ? this.core.bitfield : this.core.skipBitfield
       this.remoteBitfield.set(start, has)

lib/session-state.js

@@ -473,13 +473,37 @@ module.exports = class SessionState {
     if (this.isDefault()) {
       await storeBitfieldRange(this.storage, storage, batch.ancestors, batch.treeLength, false)
       if (batch.ancestors < this.core.header.hints.contiguousLength) {
-        storage.setHints({ contiguousLength: batch.ancestors })
+        this.core.header.hints.remoteContiguousLength = Math.min(
+          batch.length,
+          this.core.header.hints.remoteContiguousLength
+        )
+        storage.setHints({
+          contiguousLength: batch.ancestors,
+          remoteContiguousLength: this.core.header.hints.remoteContiguousLength
+        })
       }
     }
 
     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()
 
@@ -505,7 +529,10 @@ module.exports = class SessionState {
       if (this.isDefault()) {
         await storeBitfieldRange(this.storage, tx, start, end, false)
         if (start < this.core.header.hints.contiguousLength) {
-          tx.setHints({ contiguousLength: start })
+          tx.setHints({
+            contiguousLength: start,
+            remoteContiguousLength: this.core.header.hints.remoteContiguousLength
+          })
         }
       }
 
@@ -577,7 +604,10 @@ module.exports = class SessionState {
       if (this.isDefault()) {
         await storeBitfieldRange(this.storage, tx, batch.ancestors, batch.length, true)
         if (this.length === this.core.header.hints.contiguousLength) {
-          tx.setHints({ contiguousLength: this.length + values.length })
+          tx.setHints({
+            contiguousLength: this.length + values.length,
+            remoteContiguousLength: this.core.header.hints.remoteContiguousLength
+          })
         }
       }
 

test/replicate.js

@@ -2374,6 +2374,199 @@ test('hotswap works for a download with many slow peers', async function (t) {
   t.pass(`Hotswap triggered (download took ${Date.now() - start}ms)`)
 })
 
+test('remote contiguous length', async function (t) {
+  const a = await create(t)
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  a.on('remote-contiguous-length', (length) => {
+    t.is(length, 1, '`remote-contiguous-length` event fired')
+  })
+
+  replicate(a, b, t)
+
+  await b.get(0)
+
+  await eventFlush()
+
+  t.is(a.remoteContiguousLength, 1)
+})
+
+test('remote contiguous length - fully contiguous only', async function (t) {
+  t.plan(7)
+  const a = await create(t)
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a1'])
+  await a.append(['a2'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  a.on('remote-contiguous-length', (length) => {
+    t.is(length, 2, '`remote-contiguous-length` event fired')
+  })
+
+  replicate(a, b, t)
+
+  await b.get(0)
+
+  await eventFlush()
+
+  t.is(a.remoteContiguousLength, 0, 'remoteContiguousLength didnt update')
+  t.is(b.contiguousLength, 1, 'b has 1st block')
+
+  await b.get(1)
+  await eventFlush()
+
+  t.is(b.contiguousLength, 2, 'b all blocks')
+  t.is(a.remoteContiguousLength, 2, 'remoteContiguousLength updates')
+})
+
+test('remote contiguous length - updates on truncate', async function (t) {
+  const a = await create(t)
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a1'])
+  await a.append(['a2'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  replicate(a, b, t)
+
+  await b.get(0)
+  await b.get(1)
+
+  await eventFlush()
+
+  t.is(a.remoteContiguousLength, 2)
+
+  const truncateReceived = new Promise((resolve) => b.on('truncate', resolve))
+  await a.truncate(a.length - 1)
+
+  await truncateReceived
+
+  t.is(a.remoteContiguousLength, 1)
+  t.is(b.contiguousLength, 1)
+})
+
+test('remote contiguous length - event fires after truncating', async function (t) {
+  t.plan(8)
+  const a = await create(t)
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a1', 'a2', 'a3'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  replicate(a, b, t)
+
+  await b.download({ start: 0, end: a.length }).done()
+  await eventFlush() // To let `a` update `remoteContiguousLength`
+
+  t.is(a.remoteContiguousLength, 3)
+
+  const truncateReceived = new Promise((resolve) => b.on('truncate', resolve))
+  await a.truncate(1)
+
+  await truncateReceived
+
+  t.is(a.remoteContiguousLength, 1)
+  t.is(b.contiguousLength, 1)
+
+  b.on('remote-contiguous-length', () => t.pass('fires after truncating'))
+  await a.append(['a2v2', 'a3v2'])
+
+  await b.download({ start: 0, end: a.length }).done()
+  await eventFlush() // To let `a` update `remoteContiguousLength`
+
+  t.is(a.remoteContiguousLength, 3)
+  t.is(b.contiguousLength, 3)
+})
+
+test('remote contiguous length - correct after reorg', async function (t) {
+  t.plan(9)
+  const a = await create(t)
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a1', 'a2', 'a3'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  const streams = replicate(a, b, t)
+
+  await b.download({ start: 0, end: a.length }).done()
+  await eventFlush() // To let `a` update `remoteContiguousLength`
+
+  t.is(a.remoteContiguousLength, 3)
+
+  const truncateReceived = new Promise((resolve) => b.on('truncate', resolve))
+
+  // Stop replicating to clear any messages immediately after truncation & append
+  unreplicate(streams)
+
+  await a.truncate(1)
+  await a.append('a2v2')
+
+  // Rereplicate to update b
+  replicate(a, b, t)
+  await truncateReceived
+
+  t.is(a.remoteContiguousLength, 1, 'remoteContiguousLength shows truncated length')
+  t.is(b.contiguousLength, 1)
+  t.is(b.remoteContiguousLength, 2)
+
+  b.on('remote-contiguous-length', () => t.pass('fires after truncating'))
+  await a.append(['a3v2'])
+
+  await b.download({ start: 0, end: a.length }).done()
+  await eventFlush() // To let `a` update `remoteContiguousLength`
+
+  t.is(a.remoteContiguousLength, 3)
+  t.is(b.contiguousLength, 3)
+})
+
+test('remote contiguous length - persists', async function (t) {
+  const createA = await createStored(t)
+  const a = await createA()
+  await a.ready()
+  const b = await create(t, a.key)
+
+  t.is(a.remoteContiguousLength, 0)
+
+  await a.append(['a1', 'a2', 'a3'])
+
+  t.is(a.remoteContiguousLength, 0)
+
+  replicate(a, b, t)
+
+  await b.download({ start: 0, end: a.length }).done()
+  await eventFlush() // To let `a` update `remoteContiguousLength`
+
+  t.is(a.remoteContiguousLength, 3)
+
+  await a.close()
+
+  const a2 = await createA(a.key)
+  t.is(a2.remoteContiguousLength, 0, 'remoteContiguousLength initial 0 before ready')
+  await a2.ready()
+  t.teardown(() => a2.close())
+
+  t.is(a2.remoteContiguousLength, 3)
+})
+
 async function createAndDownload(t, core) {
   const b = await create(t, core.key)
   replicate(core, b, t, { teardown: false })