Cache score by decayInterval

Author: twoeths

test/peer-score.spec.js

@@ -1,8 +1,10 @@
+const sinon = require('sinon')
 const { expect } = require('chai')
 const PeerId = require('peer-id')
 const delay = require('delay')
 
 const { PeerScore, createPeerScoreParams, createTopicScoreParams } = require('../src/score')
+const computeScoreModule = require('../src/score/compute-score')
 const { ERR_TOPIC_VALIDATOR_IGNORE, ERR_TOPIC_VALIDATOR_REJECT } = require('../src/constants')
 const { makeTestMessage, getMsgId } = require('./utils')
 
@@ -642,7 +644,6 @@ describe('PeerScore', () => {
     const ps = new PeerScore(params, connectionManager, getMsgId)
     ps.addPeer(peerA)
     ps.graft(peerA, mytopic)
-    
     // score should equal -1000 (app-specific score)
     const expected = -1000
     ps._refreshScores()
@@ -665,3 +666,70 @@ describe('PeerScore', () => {
     expect(aScore).to.equal(0)
   })
 })
+
+describe('PeerScore score cache', function () {
+  let ps2
+  let peerA
+  const sandbox = sinon.createSandbox()
+  let computeStoreStub
+  const params = createPeerScoreParams({
+    appSpecificScore: () => -1000,
+    appSpecificWeight: 1,
+    retainScore: 800,
+    decayInterval: 1000,
+    topics: {a: {topicWeight: 10}}
+  })
+
+  beforeEach(async () => {
+    sandbox.useFakeTimers()
+    peerA = (await PeerId.create({keyType: 'secp256k1'})).toB58String()
+    computeStoreStub = sandbox.stub(computeScoreModule, 'computeScore')
+    ps2 = new PeerScore(params, connectionManager, getMsgId)
+  })
+
+  afterEach(() => {
+    sandbox.restore()
+  })
+
+  it('should compute first time', function () {
+    computeStoreStub.returns(10)
+    ps2.addPeer(peerA)
+    expect(computeStoreStub.calledOnce).to.be.false
+    ps2.score(peerA)
+    expect(computeStoreStub.calledOnce).to.be.true
+    // this time peerA score is cached
+    ps2.score(peerA)
+    expect(computeStoreStub.calledOnce).to.be.true
+  })
+
+  const testCases = [
+    {name: 'decayInterval timeout', fun: () => sandbox.clock.tick(params.decayInterval)},
+    {name: '_refreshScores', fun: () => ps2._refreshScores()},
+    {name: 'addPenalty', fun: () => ps2.addPenalty(peerA, 10)},
+    {name: 'graft', fun: () => ps2.graft(peerA, 'a')},
+    {name: 'prune', fun: () => ps2.prune(peerA, 'a')},
+    {name: '_markInvalidMessageDelivery', fun: () => ps2._markInvalidMessageDelivery(peerA, {topicIDs: ['a']})},
+    {name: '_markFirstMessageDelivery', fun: () => ps2._markFirstMessageDelivery(peerA, {topicIDs: ['a']})},
+    {name: '_markDuplicateMessageDelivery', fun: () => ps2._markDuplicateMessageDelivery(peerA, {topicIDs: ['a']})},
+    {name: '_setIPs', fun: () => ps2._setIPs(peerA, [], ['127.0.0.1'])},
+    {name: '_removeIPs', fun: () => ps2._removeIPs(peerA, ['127.0.0.1'])},
+    {name: '_updateIPs', fun: () => ps2._updateIPs()},
+  ]
+
+  for (const {name, fun} of testCases) {
+    it(`should invalidate the cache after ${name}`, function () {
+      computeStoreStub.returns(10)
+      ps2.addPeer(peerA)
+      ps2.score(peerA)
+      expect(computeStoreStub.calledOnce).to.be.true
+      // the score is cached
+      ps2.score(peerA)
+      expect(computeStoreStub.calledOnce).to.be.true
+      // invalidate the cache
+      fun()
+      // should not use the cache
+      ps2.score(peerA)
+      expect(computeStoreStub.calledTwice).to.be.true
+    })
+  }
+})

ts/score/peer-score.ts

@@ -30,6 +30,15 @@ export class PeerScore {
    * IP colocation tracking; maps IP => set of peers.
    */
   peerIPs: Map<string, Set<string>>
+  scoreCache: Map<string, number>
+  /**
+   * Flag to mark a peer score cache valid or not.
+   */
+  scoreCacheValid: Map<string, boolean>
+  /**
+   * The last time the score for a peer was cached.
+   */
+  scoreCacheTime: Map<string, number>
   /**
    * Recent message delivery timing/participants
    */
@@ -47,6 +56,9 @@ export class PeerScore {
     this._connectionManager = connectionManager
     this.peerStats = new Map()
     this.peerIPs = new Map()
+    this.scoreCache = new Map()
+    this.scoreCacheValid = new Map()
+    this.scoreCacheTime = new Map()
     this.deliveryRecords = new MessageDeliveries()
     this.msgId = msgId
   }
@@ -153,6 +165,8 @@ export class PeerScore {
       if (pstats.behaviourPenalty < decayToZero) {
         pstats.behaviourPenalty = 0
       }
+
+      this.scoreCacheValid.set(id, false)
     })
   }
 
@@ -166,7 +180,18 @@ export class PeerScore {
     if (!pstats) {
       return 0
     }
-    return computeScore(id, pstats, this.params, this.peerIPs)
+
+    const now = Date.now()
+    if (this.scoreCacheValid.get(id) && now - (this.scoreCacheTime.get(id) ?? 0) < this.params.decayInterval) {
+      const score = this.scoreCache.get(id)
+      if (score !== undefined) return score
+    }
+
+    const score = computeScore(id, pstats, this.params, this.peerIPs)
+    this.scoreCacheValid.set(id, true)
+    this.scoreCacheTime.set(id, now)
+    this.scoreCache.set(id, score)
+    return score
   }
 
   /**
@@ -181,6 +206,7 @@ export class PeerScore {
       return
     }
     pstats.behaviourPenalty += penalty
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -199,6 +225,10 @@ export class PeerScore {
     const ips = this._getIPs(id)
     this._setIPs(id, ips, pstats.ips)
     pstats.ips = ips
+
+    // initialize score cache
+    this.scoreCacheTime.set(id, 0)
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -219,6 +249,11 @@ export class PeerScore {
       return
     }
 
+    // delete score cache
+    this.scoreCache.delete(id)
+    this.scoreCacheTime.delete(id)
+    this.scoreCacheValid.delete(id)
+
     // furthermore, when we decide to retain the score, the firstMessageDelivery counters are
     // reset to 0 and mesh delivery penalties applied.
     Object.entries(pstats.topics).forEach(([topic, tstats]) => {
@@ -257,6 +292,7 @@ export class PeerScore {
     tstats.graftTime = Date.now()
     tstats.meshTime = 0
     tstats.meshMessageDeliveriesActive = false
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -282,6 +318,7 @@ export class PeerScore {
       tstats.meshFailurePenalty += deficit * deficit
     }
     tstats.inMesh = false
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -416,6 +453,7 @@ export class PeerScore {
 
       tstats.invalidMessageDeliveries += 1
     })
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -453,6 +491,7 @@ export class PeerScore {
         tstats.meshMessageDeliveries = cap
       }
     })
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -496,6 +535,7 @@ export class PeerScore {
         tstats.meshMessageDeliveries = cap
       }
     })
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -556,6 +596,8 @@ export class PeerScore {
         this.peerIPs.delete(ip)
       }
     }
+
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -576,6 +618,8 @@ export class PeerScore {
         this.peerIPs.delete(ip)
       }
     })
+
+    this.scoreCacheValid.set(id, false)
   }
 
   /**
@@ -587,6 +631,7 @@ export class PeerScore {
       const newIPs = this._getIPs(id)
       this._setIPs(id, newIPs, pstats.ips)
       pstats.ips = newIPs
+      this.scoreCacheValid.set(id, false)
     })
   }
 }