Bugfix: send the broadcast to all peers (+test)

Author: HDegroote

lib/replicator.js

@@ -382,6 +382,7 @@ class Peer {
     this.remoteSupportsSeeks = false
     this.inflightRange = inflightRange
     this.remoteSegmentsWanted = new Set()
+    this.fullyDownloaded = false
 
     this.paused = false
     this.removed = false
@@ -482,6 +483,7 @@ class Peer {
   }
 
   signalUpgrade () {
+    if (!this.replicator.hasFullCore && this.fullyDownloaded) this.fullyDownloaded = false
     if (this._shouldUpdateCanUpgrade() === true) this._updateCanUpgradeAndSync()
     else this.sendSync()
   }
@@ -508,7 +510,7 @@ class Peer {
 
       // Always send the broadcast when we switch from sparse to fully contiguous
       if (!this.fullyDownloaded) {
-        this.replicator.fullyDownloaded = true
+        this.fullyDownloaded = true
         force = true
       }
     }
@@ -1509,7 +1511,6 @@ module.exports = class Replicator {
     this.destroyed = false
     this.downloading = false
     this.activeSessions = 0
-    this.fullyDownloaded = false
 
     this.hotswaps = new HotswapQueue()
     this.inflightRange = inflightRange || DEFAULT_MAX_INFLIGHT
@@ -1643,8 +1644,6 @@ module.exports = class Replicator {
     if (!this._blocks.isEmpty() || this._ranges.length !== 0 || this._seeks.length !== 0) {
       this._updateNonPrimary(true)
     }
-
-    if (!this.hasFullCore && this.fullyDownloaded) this.fullyDownloaded = false
   }
 
   // Called externally when a conflict has been detected and verified

test/replicate.js

@@ -2164,6 +2164,58 @@ test('range is broadcast when a core is fully available', async function (t) {
   t.is(writer.replicator.peers[0]._remoteContiguousLength, 7, 'writer detected reader is fully contig')
 })
 
+test('range is broadcast when a core is fully available (multiple peers)', async function (t) {
+  const writer = await create(t)
+
+  await writer.append(['a', 'b', 'c'])
+  const reader1 = await create(t, writer.key)
+  const reader2 = await create(t, writer.key)
+
+  replicate(writer, reader1, t)
+  await reader1.get(0)
+  const writerToReader1 = writer.replicator.peers[0]
+  const reader1ToWriter = reader1.replicator.peers[0]
+
+  replicate(writer, reader2, t)
+  await reader2.get(0)
+  const writerToReader2 = writer.replicator.peers[1]
+  const reader2ToWriter = reader1.replicator.peers[0]
+
+  replicate(reader1, reader2, t)
+  // Give time for replication to add the peers
+  await new Promise(resolve => setTimeout(resolve, 100))
+
+  const reader1ToReader2 = reader1.replicator.peers[1]
+  const reader2ToReader1 = reader2.replicator.peers[1]
+  t.is(reader1ToReader2 !== undefined, true, 'sanity check')
+  t.is(reader2ToReader1 !== undefined, true, 'sanity check')
+
+  await new Promise(resolve => setTimeout(resolve, 100))
+
+  t.is(writerToReader1._remoteContiguousLength, 0, 'reader1 skipped range update to writer (not contig yet)')
+  t.is(writerToReader2._remoteContiguousLength, 0, 'reader2 skipped range update to writer (not contig yet)')
+  t.is(reader1ToWriter._remoteContiguousLength, 3, 'writer updated for reader1')
+  t.is(reader2ToWriter._remoteContiguousLength, 3, 'writer updated for reader2')
+
+  await reader1.get(1)
+  await reader1.get(2)
+  await reader2.get(1)
+  await reader2.get(2)
+  await new Promise(resolve => setTimeout(resolve, 100))
+
+  t.is(reader1.contiguousLength, 3, 'sanity check')
+  t.is(reader2.contiguousLength, 3, 'sanity check')
+  t.is(writerToReader1._remoteContiguousLength, 3, 'reader1 sent range update to writer (became contig)')
+  t.is(writerToReader2._remoteContiguousLength, 3, 'reader2 sent range update to writer (became contig)')
+  t.is(reader1ToReader2._remoteContiguousLength, 3, 'reader2 broadcast to reader1 too')
+  t.is(reader2ToReader1._remoteContiguousLength, 3, 'reader1 broadcast to reader2 too')
+
+  await writer.append(['d', 'e'])
+  await new Promise(resolve => setTimeout(resolve, 100))
+  t.is(reader1ToWriter._remoteContiguousLength, 5, 'writer updated for reader1')
+  t.is(reader2ToWriter._remoteContiguousLength, 5, 'writer updated for reader2')
+})
+
 async function waitForRequestBlock (core) {
   while (true) {
     const reqBlock = core.core.replicator._inflight._requests.find(req => req && req.block)