feat: expire peerstore data (#3019)

Expires peer store multiaddrs after an hour and removes peer store
peers after six hours if their addresses have not been updated.

Fixes #3017

Author: achingbrain

packages/peer-store/src/constants.ts

@@ -0,0 +1,2 @@
+export const MAX_ADDRESS_AGE = 3_600_000
+export const MAX_PEER_AGE = 21_600_000

packages/peer-store/src/index.ts

@@ -27,7 +27,31 @@ export interface AddressFilter {
 }
 
 export interface PersistentPeerStoreInit {
+  /**
+   * Used to remove multiaddrs of peers before storing them. The default is to
+   * store all addresses
+   */
   addressFilter?: AddressFilter
+
+  /**
+   * The multiaddrs for a given peer will expire after this number of ms after
+   * which they must be re-fetched using the peer routing.
+   *
+   * Defaults to one hour.
+   *
+   * @default 3_600_000
+   */
+  maxAddressAge?: number
+
+  /**
+   * Any peer without multiaddrs that has not been updated after this number of
+   * ms will be evicted from the peer store.
+   *
+   * Defaults to six hours.
+   *
+   * @default 21_600_000
+   */
+  maxPeerAge?: number
 }
 
 /**

packages/peer-store/src/pb/peer.proto

@@ -18,6 +18,9 @@ message Peer {
 
   // Any tags the peer has
   map<string, Tag> tags = 7;
+
+  // timestamp in ms of when this peer was last updated
+  optional uint64 updated = 8 [jstype = JS_NUMBER];
 }
 
 // Address represents a single multiaddr
@@ -26,6 +29,9 @@ message Address {
 
   // Flag to indicate if the address comes from a certified source
   optional bool isCertified = 2;
+
+  // timestamp in ms of when this address was observed
+  optional uint64 observed = 3 [jstype = JS_NUMBER];
 }
 
 message Tag {

packages/peer-store/src/pb/peer.ts

@@ -15,6 +15,7 @@ export interface Peer {
   peerRecordEnvelope?: Uint8Array
   metadata: Map<string, Uint8Array>
   tags: Map<string, Tag>
+  updated?: number
 }
 
 export namespace Peer {
@@ -208,6 +209,11 @@ export namespace Peer {
           }
         }
 
+        if (obj.updated != null) {
+          w.uint32(64)
+          w.uint64Number(obj.updated)
+        }
+
         if (opts.lengthDelimited !== false) {
           w.ldelim()
         }
@@ -273,6 +279,10 @@ export namespace Peer {
               obj.tags.set(entry.key, entry.value)
               break
             }
+            case 8: {
+              obj.updated = reader.uint64Number()
+              break
+            }
             default: {
               reader.skipType(tag & 7)
               break
@@ -299,6 +309,7 @@ export namespace Peer {
 export interface Address {
   multiaddr: Uint8Array
   isCertified?: boolean
+  observed?: number
 }
 
 export namespace Address {
@@ -321,6 +332,11 @@ export namespace Address {
           w.bool(obj.isCertified)
         }
 
+        if (obj.observed != null) {
+          w.uint32(24)
+          w.uint64Number(obj.observed)
+        }
+
         if (opts.lengthDelimited !== false) {
           w.ldelim()
         }
@@ -343,6 +359,10 @@ export namespace Address {
               obj.isCertified = reader.bool()
               break
             }
+            case 3: {
+              obj.observed = reader.uint64Number()
+              break
+            }
             default: {
               reader.skipType(tag & 7)
               break

packages/peer-store/src/store.ts

@@ -1,11 +1,12 @@
-import { InvalidParametersError } from '@libp2p/interface'
+import { NotFoundError } from '@libp2p/interface'
 import { peerIdFromCID } from '@libp2p/peer-id'
 import mortice, { type Mortice } from 'mortice'
 import { base32 } from 'multiformats/bases/base32'
 import { CID } from 'multiformats/cid'
-import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
+import { MAX_ADDRESS_AGE, MAX_PEER_AGE } from './constants.js'
 import { Peer as PeerPB } from './pb/peer.js'
-import { bytesToPeer } from './utils/bytes-to-peer.js'
+import { bytesToPeer, pbToPeer } from './utils/bytes-to-peer.js'
+import { peerEquals } from './utils/peer-equals.js'
 import { NAMESPACE_COMMON, peerIdToDatastoreKey } from './utils/peer-id-to-datastore-key.js'
 import { toPeerPB } from './utils/to-peer-pb.js'
 import type { AddressFilter, PersistentPeerStoreComponents, PersistentPeerStoreInit } from './index.js'
@@ -19,27 +20,33 @@ export interface PeerUpdate extends PeerUpdateExternal {
   updated: boolean
 }
 
-function decodePeer (key: Key, value: Uint8Array): Peer {
+export interface ExistingPeer {
+  peerPB: PeerPB
+  peer: Peer
+}
+
+function keyToPeerId (key: Key): PeerId {
   // /peers/${peer-id-as-libp2p-key-cid-string-in-base-32}
   const base32Str = key.toString().split('/')[2]
   const buf = CID.parse(base32Str, base32)
-  const peerId = peerIdFromCID(buf)
 
-  return bytesToPeer(peerId, value)
+  return peerIdFromCID(buf)
 }
 
-function mapQuery (query: PeerQuery): Query {
-  if (query == null) {
-    return {}
-  }
+function decodePeer (key: Key, value: Uint8Array, maxAddressAge: number): Peer {
+  const peerId = keyToPeerId(key)
+
+  return bytesToPeer(peerId, value, maxAddressAge)
+}
 
+function mapQuery (query: PeerQuery, maxAddressAge: number): Query {
   return {
     prefix: NAMESPACE_COMMON,
     filters: (query.filters ?? []).map(fn => ({ key, value }) => {
-      return fn(decodePeer(key, value))
+      return fn(decodePeer(key, value, maxAddressAge))
     }),
     orders: (query.orders ?? []).map(fn => (a, b) => {
-      return fn(decodePeer(a.key, a.value), decodePeer(b.key, b.value))
+      return fn(decodePeer(a.key, a.value, maxAddressAge), decodePeer(b.key, b.value, maxAddressAge))
     })
   }
 }
@@ -50,6 +57,8 @@ export class PersistentStore {
   public readonly lock: Mortice
   private readonly addressFilter?: AddressFilter
   private readonly log: Logger
+  private readonly maxAddressAge: number
+  private readonly maxPeerAge: number
 
   constructor (components: PersistentPeerStoreComponents, init: PersistentPeerStoreInit = {}) {
     this.log = components.logger.forComponent('libp2p:peer-store')
@@ -60,115 +69,146 @@ export class PersistentStore {
       name: 'peer-store',
       singleProcess: true
     })
+    this.maxAddressAge = init.maxAddressAge ?? MAX_ADDRESS_AGE
+    this.maxPeerAge = init.maxPeerAge ?? MAX_PEER_AGE
   }
 
   async has (peerId: PeerId): Promise<boolean> {
-    return this.datastore.has(peerIdToDatastoreKey(peerId))
+    try {
+      await this.load(peerId)
+
+      return true
+    } catch (err: any) {
+      if (err.name !== 'NotFoundError') {
+        throw err
+      }
+    }
+
+    return false
   }
 
   async delete (peerId: PeerId): Promise<void> {
     if (this.peerId.equals(peerId)) {
-      throw new InvalidParametersError('Cannot delete self peer')
+      return
     }
 
     await this.datastore.delete(peerIdToDatastoreKey(peerId))
   }
 
   async load (peerId: PeerId): Promise<Peer> {
-    const buf = await this.datastore.get(peerIdToDatastoreKey(peerId))
+    const key = peerIdToDatastoreKey(peerId)
+    const buf = await this.datastore.get(key)
+    const peer = PeerPB.decode(buf)
+
+    if (this.#peerIsExpired(peer)) {
+      await this.datastore.delete(key)
+      throw new NotFoundError()
+    }
 
-    return bytesToPeer(peerId, buf)
+    return pbToPeer(peerId, peer, this.maxAddressAge)
   }
 
   async save (peerId: PeerId, data: PeerData): Promise<PeerUpdate> {
-    const {
-      existingBuf,
-      existingPeer
-    } = await this.#findExistingPeer(peerId)
+    const existingPeer = await this.#findExistingPeer(peerId)
 
     const peerPb: PeerPB = await toPeerPB(peerId, data, 'patch', {
       addressFilter: this.addressFilter
     })
 
-    return this.#saveIfDifferent(peerId, peerPb, existingBuf, existingPeer)
+    return this.#saveIfDifferent(peerId, peerPb, existingPeer)
   }
 
   async patch (peerId: PeerId, data: Partial<PeerData>): Promise<PeerUpdate> {
-    const {
-      existingBuf,
-      existingPeer
-    } = await this.#findExistingPeer(peerId)
+    const existingPeer = await this.#findExistingPeer(peerId)
 
     const peerPb: PeerPB = await toPeerPB(peerId, data, 'patch', {
       addressFilter: this.addressFilter,
       existingPeer
     })
 
-    return this.#saveIfDifferent(peerId, peerPb, existingBuf, existingPeer)
+    return this.#saveIfDifferent(peerId, peerPb, existingPeer)
   }
 
   async merge (peerId: PeerId, data: PeerData): Promise<PeerUpdate> {
-    const {
-      existingBuf,
-      existingPeer
-    } = await this.#findExistingPeer(peerId)
+    const existingPeer = await this.#findExistingPeer(peerId)
 
     const peerPb: PeerPB = await toPeerPB(peerId, data, 'merge', {
       addressFilter: this.addressFilter,
       existingPeer
     })
 
-    return this.#saveIfDifferent(peerId, peerPb, existingBuf, existingPeer)
+    return this.#saveIfDifferent(peerId, peerPb, existingPeer)
   }
 
   async * all (query?: PeerQuery): AsyncGenerator<Peer, void, unknown> {
-    for await (const { key, value } of this.datastore.query(mapQuery(query ?? {}))) {
-      const peer = decodePeer(key, value)
+    for await (const { key, value } of this.datastore.query(mapQuery(query ?? {}, this.maxAddressAge))) {
+      const peerId = keyToPeerId(key)
 
-      if (peer.id.equals(this.peerId)) {
-        // Skip self peer if present
+      // skip self peer if present
+      if (peerId.equals(this.peerId)) {
         continue
       }
 
-      yield peer
+      const peer = PeerPB.decode(value)
+
+      // remove expired peer
+      if (this.#peerIsExpired(peer)) {
+        await this.datastore.delete(key)
+        continue
+      }
+
+      yield pbToPeer(peerId, peer, this.maxAddressAge)
     }
   }
 
-  async #findExistingPeer (peerId: PeerId): Promise<{ existingBuf?: Uint8Array, existingPeer?: Peer }> {
+  async #findExistingPeer (peerId: PeerId): Promise<ExistingPeer | undefined> {
     try {
-      const existingBuf = await this.datastore.get(peerIdToDatastoreKey(peerId))
-      const existingPeer = bytesToPeer(peerId, existingBuf)
+      const key = peerIdToDatastoreKey(peerId)
+      const buf = await this.datastore.get(key)
+      const peerPB = PeerPB.decode(buf)
+
+      // remove expired peer
+      if (this.#peerIsExpired(peerPB)) {
+        await this.datastore.delete(key)
+        throw new NotFoundError()
+      }
 
       return {
-        existingBuf,
-        existingPeer
+        peerPB,
+        peer: bytesToPeer(peerId, buf, this.maxAddressAge)
       }
     } catch (err: any) {
       if (err.name !== 'NotFoundError') {
         this.log.error('invalid peer data found in peer store - %e', err)
       }
     }
-
-    return {}
   }
 
-  async #saveIfDifferent (peerId: PeerId, peer: PeerPB, existingBuf?: Uint8Array, existingPeer?: Peer): Promise<PeerUpdate> {
+  async #saveIfDifferent (peerId: PeerId, peer: PeerPB, existingPeer?: ExistingPeer): Promise<PeerUpdate> {
+    // record last update
+    peer.updated = Date.now()
     const buf = PeerPB.encode(peer)
 
-    if (existingBuf != null && uint8ArrayEquals(buf, existingBuf)) {
-      return {
-        peer: bytesToPeer(peerId, buf),
-        previous: existingPeer,
-        updated: false
-      }
-    }
-
     await this.datastore.put(peerIdToDatastoreKey(peerId), buf)
 
     return {
-      peer: bytesToPeer(peerId, buf),
-      previous: existingPeer,
-      updated: true
+      peer: bytesToPeer(peerId, buf, this.maxAddressAge),
+      previous: existingPeer?.peer,
+      updated: existingPeer == null || !peerEquals(peer, existingPeer.peerPB)
+    }
+  }
+
+  #peerIsExpired (peer: PeerPB): boolean {
+    if (peer.updated == null) {
+      return true
     }
+
+    const expired = peer.updated < (Date.now() - this.maxPeerAge)
+    const minAddressObserved = Date.now() - this.maxAddressAge
+    const addrs = peer.addresses.filter(addr => {
+      return addr.observed != null && addr.observed > minAddressObserved
+    })
+
+    return expired && addrs.length === 0
   }
 }