feat: add iwant request tracking

test/peerScore.spec.js

@@ -2,20 +2,13 @@ const { expect } = require('chai')
 const PeerId = require('peer-id')
 const { utils } = require('libp2p-pubsub')
 const delay = require('delay')
+
 const { PeerScore, createPeerScoreParams, createTopicScoreParams } = require('../src/score')
+const { makeTestMessage } = require('./utils')
 
 const connectionManager = new Map()
 connectionManager.getAll = () => ([])
 
-const makeTestMessage = (i, topicIDs = []) => {
-  return {
-    seqno: Buffer.alloc(8, i),
-    data: Buffer.from([i]),
-    from: "test",
-    topicIDs
-  }
-}
-
 describe('PeerScore', () => {
   it('should score based on time in mesh', async () => {
     // Create parameters with reasonable default values

test/tracer.spec.js

@@ -0,0 +1,68 @@
+const { expect } = require('chai')
+const delay = require('delay')
+const { utils } = require('libp2p-pubsub')
+
+const { IWantTracer } = require('../src/tracer')
+const constants = require('../src/constants')
+const { makeTestMessage } = require('./utils')
+
+const getMsgId = (msg) => utils.msgId(msg.from, msg.seqno)
+
+describe('IWantTracer', () => {
+  it('should track broken promises', async function () {
+    // tests that unfullfilled promises are tracked correctly
+    this.timeout(6000)
+    const t = new IWantTracer(getMsgId)
+    const peerA = 'A'
+    const peerB = 'B'
+
+    const msgIds = []
+    for (let i = 0; i < 100; i++) {
+      const m = makeTestMessage(i)
+      m.from = Buffer.from(peerA)
+      msgIds.push(getMsgId(m))
+    }
+
+    t.addPromise(peerA, msgIds)
+    t.addPromise(peerB, msgIds)
+
+    // no broken promises yet
+    let brokenPromises = t.getBrokenPromises()
+    expect(brokenPromises.size).to.be.equal(0)
+
+    // make promises break
+    await delay(constants.GossipsubIWantFollowupTime + 10)
+
+    brokenPromises = t.getBrokenPromises()
+    expect(brokenPromises.size).to.be.equal(2)
+    expect(brokenPromises.get(peerA)).to.be.equal(1)
+    expect(brokenPromises.get(peerB)).to.be.equal(1)
+  })
+  it('should track unbroken promises', async function () {
+    // like above, but this time we deliver messages to fullfil the promises
+    this.timeout(6000)
+    const t = new IWantTracer(getMsgId)
+    const peerA = 'A'
+    const peerB = 'B'
+
+    const msgs = []
+    const msgIds = []
+    for (let i = 0; i < 100; i++) {
+      const m = makeTestMessage(i)
+      m.from = Buffer.from(peerA)
+      msgs.push(m)
+      msgIds.push(getMsgId(m))
+    }
+
+    t.addPromise(peerA, msgIds)
+    t.addPromise(peerB, msgIds)
+
+    msgs.forEach(msg => t.deliverMessage(peerA, msg))
+
+    await delay(constants.GossipsubIWantFollowupTime + 10)
+
+    // there should be no broken promises
+    const brokenPromises = t.getBrokenPromises()
+    expect(brokenPromises.size).to.be.equal(0)
+  })
+})

test/utils/index.js

@@ -40,7 +40,8 @@ exports.createFloodsubNode = createFloodsubNode
 
 for (const [k, v] of Object.entries({
   ...require('./createPeer'),
-  ...require('./createGossipsub')
+  ...require('./createGossipsub'),
+  ...require('./makeTestMessage')
 })) {
   exports[k] = v
 }

test/utils/makeTestMessage.js

@@ -0,0 +1,10 @@
+const makeTestMessage = (i, topicIDs = []) => {
+  return {
+    seqno: Buffer.alloc(8, i),
+    data: Buffer.from([i]),
+    from: "test",
+    topicIDs
+  }
+}
+
+module.exports.makeTestMessage = makeTestMessage

ts/heartbeat.ts

@@ -94,6 +94,9 @@ export class Heartbeat {
     this.gossipsub.peerhave.clear()
     this.gossipsub.iasked.clear()
 
+    // apply IWANT request penalties
+    this.gossipsub._applyIwantPenalties()
+
     // ensure direct peers are connected
     this.gossipsub._directConnect()
 

ts/index.ts

@@ -14,6 +14,7 @@ import { getGossipPeers } from './getGossipPeers'
 import { createGossipRpc, shuffle, hasGossipProtocol } from './utils'
 import { Peer } from './peer'
 import { PeerScore, PeerScoreParams, PeerScoreThresholds, createPeerScoreParams, createPeerScoreThresholds } from './score'
+import { IWantTracer } from './tracer'
 import { AddrInfo, Libp2p } from './interfaces'
 // @ts-ignore
 import TimeCache = require('time-cache')
@@ -52,6 +53,7 @@ class Gossipsub extends BasicPubsub {
   outbound: Map<Peer, boolean>
   score: PeerScore
   heartbeatTicks: number
+  gossipTracer: IWantTracer
   _libp2p: Libp2p
   _options: GossipOptions
 
@@ -197,6 +199,11 @@ class Gossipsub extends BasicPubsub {
      */
     this.heartbeatTicks = 0
 
+    /**
+     * Tracks IHAVE/IWANT promises broken by peers
+     */
+    this.gossipTracer = new IWantTracer(this._msgIdFn)
+
     /**
      * libp2p
      */
@@ -341,7 +348,9 @@ class Gossipsub extends BasicPubsub {
    * @returns {void}
    */
   _publishFrom (peer: Peer, msg: InMessage): void {
-    this.score.deliverMessage(peer.id.toB58String(), msg as Message)
+    const id = peer.id.toB58String()
+    this.score.deliverMessage(id, msg as Message)
+    this.gossipTracer.deliverMessage(id, msg as Message)
     const topics = msg.topicIDs
 
     // If options.gossipIncoming is false, do NOT emit incoming messages to peers
@@ -401,15 +410,18 @@ class Gossipsub extends BasicPubsub {
    */
   _processTopicValidatorResult (topic: string, peer: Peer, message: Message, result: unknown): boolean {
     if (typeof result === 'string') {
+      const id = peer.id.toB58String()
       // assume an extended topic validator result
       switch (result) {
         case ExtendedValidatorResult.accept:
           return true
         case ExtendedValidatorResult.reject:
-          this.score.rejectMessage(peer.id.toB58String(), message)
+          this.score.rejectMessage(id, message)
+          this.gossipTracer.rejectMessage(id, message)
           return false
         case ExtendedValidatorResult.ignore:
-          this.score.ignoreMessage(peer.id.toB58String(), message)
+          this.score.ignoreMessage(id, message)
+          this.gossipTracer.rejectMessage(id, message)
           return false
       }
     }
@@ -492,6 +504,8 @@ class Gossipsub extends BasicPubsub {
     iwantList = iwantList.slice(0, iask)
     this.iasked.set(id, iasked + iask)
 
+    this.gossipTracer.addPromise(id, iwantList)
+
     return {
       messageIDs: iwantList
     }
@@ -671,6 +685,17 @@ class Gossipsub extends BasicPubsub {
     }
   }
 
+  /**
+   * Apply penalties from broken IHAVE/IWANT promises
+   * @returns {void}
+   */
+  _applyIwantPenalties (): void {
+    this.gossipTracer.getBrokenPromises().forEach((count, p) => {
+      this.log('peer %s didn\'t follow up in %d IWANT requests; adding penalty', p, count)
+      this.score.addPenalty(p, count)
+    })
+  }
+
   /**
    * Clear expired backoff expiries
    * @returns {void}
@@ -756,6 +781,7 @@ class Gossipsub extends BasicPubsub {
     this.iasked = new Map()
     this.backoff = new Map()
     this.outbound = new Map()
+    this.gossipTracer.clear()
     clearTimeout(this._directPeerInitial)
   }
 

ts/tracer.ts

@@ -0,0 +1,99 @@
+import { Message } from './message'
+import { GossipsubIWantFollowupTime } from './constants'
+
+/**
+ * IWantTracer is an internal tracer that tracks IWANT requests in order to penalize
+ * peers who don't follow up on IWANT requests after an IHAVE advertisement.
+ * The tracking of promises is probabilistic to avoid using too much memory.
+ *
+ * Note: Do not confuse these 'promises' with JS Promise objects.
+ * These 'promises' are merely expectations of a peer's behavior.
+ */
+export class IWantTracer {
+  getMsgId: (msg: Message) => string
+  /**
+   * Promises to deliver a message
+   * Map per message id, per peer, promise expiration time
+   */
+  promises: Map<string, Map<string, number>>
+  constructor (getMsgId: (msg: Message) => string) {
+    this.getMsgId = getMsgId
+    this.promises = new Map()
+  }
+
+  /**
+   * Track a promise to deliver a message from a list of msgIDs we are requesting
+   * @param {string} p peer id
+   * @param {string[]} msgIds
+   * @returns {void}
+   */
+  addPromise (p: string, msgIds: string[]): void {
+    // pick msgId randomly from the list
+    const ix = Math.floor(Math.random() * msgIds.length)
+    const msgId = msgIds[ix]
+
+    let peers = this.promises.get(msgId)
+    if (!peers) {
+      peers = new Map()
+      this.promises.set(msgId, peers)
+    }
+
+    if (!peers.has(p)) {
+      peers.set(p, Date.now() + GossipsubIWantFollowupTime)
+    }
+  }
+
+  /**
+   * Returns the number of broken promises for each peer who didn't follow up on an IWANT request.
+   * @returns {Map<string, number>}
+   */
+  getBrokenPromises (): Map<string, number> {
+    const now = Date.now()
+    const result = new Map<string, number>()
+
+    this.promises.forEach((peers, msgId) => {
+      peers.forEach((expire, p) => {
+        // the promise has been broken
+        if (expire < now) {
+          // add 1 to result
+          result.set(p, (result.get(p) || 0) + 1)
+          // delete from tracked promises
+          peers.delete(p)
+        }
+      })
+      // clean up empty promises for a msgId
+      if (!peers.size) {
+        this.promises.delete(msgId)
+      }
+    })
+
+    return result
+  }
+
+  /**
+   * Someone delivered a message, stop tracking promises for it
+   * @param {string} p peer id
+   * @param {Message} msg
+   * @returns {void}
+   */
+  deliverMessage (p: string, msg: Message): void {
+    const msgId = this.getMsgId(msg)
+    this.promises.delete(msgId)
+  }
+
+  /**
+   * A message got rejected, so we can stop tracking promises and let the score penalty apply from invalid message delivery,
+   * unless its an obviously invalid message.
+   * @param {string} p peer id
+   * @param {Message} msg
+   * @returns {void}
+   */
+  rejectMessage (p: string, msg: Message): void {
+    const msgId = this.getMsgId(msg)
+    this.promises.delete(msgId)
+  }
+
+  clear (): void {
+    this.promises.clear()
+  }
+}