Add core.remoteContiguousLength & remote-contiguous-length event

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

@@ -561,6 +561,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

@@ -197,7 +197,8 @@ module.exports = class Core {
         },
         hints: {
           reorgs: [],
-          contiguousLength: 0
+          contiguousLength: 0,
+          remoteContiguousLength: 0
         }
       }
 
@@ -281,7 +282,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()
     }
 
@@ -391,6 +392,17 @@ module.exports = class Core {
     return this.state.flushed()
   }
 
+  async updateRemoteContiguousLength(length) {
+    if (length <= this.header.hints.remoteContiguousLength) return
+    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)
+    }
+  }
+
   async _validateCommit(state, treeLength) {
     if (this.state.length > state.length) {
       return false // TODO: partial commit and truncation possible in the future
@@ -852,7 +864,8 @@ function parseHeader(info) {
     tree: info.head || getDefaultTree(),
     hints: {
       reorgs: [],
-      contiguousLength: info.hints ? info.hints.contiguousLength : 0
+      contiguousLength: info.hints?.contiguousLength ?? 0,
+      remoteContiguousLength: info.hints?.remoteContiguousLength ?? 0
     }
   }
 }

lib/replicator.js

@@ -513,18 +513,18 @@ 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
 
@@ -1149,15 +1149,26 @@ 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) {
+          try {
+            await this.core.updateRemoteContiguousLength(length)
+          } catch (err) {
+            safetyCatch(err)
+            this.paused = true
+            return
+          }
+        }
+      }
     } 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 })