feat: add iwant request tracking

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.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()
+  }
+}