fix: use a per-peer lock (#3162)

Instead of locking the whole peer store on a multi-read/single-write
basis, make the lock more granular and use a per-peer-id lock.

Author: achingbrain

packages/peer-store/package.json

@@ -50,6 +50,7 @@
   "dependencies": {
     "@libp2p/crypto": "^5.1.4",
     "@libp2p/interface": "^2.10.2",
+    "@libp2p/peer-collections": "^6.0.30",
     "@libp2p/peer-id": "^5.1.5",
     "@libp2p/peer-record": "^8.0.30",
     "@multiformats/multiaddr": "^12.4.0",

packages/peer-store/src/index.ts

@@ -10,7 +10,7 @@ import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record'
 import all from 'it-all'
 import { PersistentStore } from './store.js'
 import type { PeerUpdate } from './store.js'
-import type { ComponentLogger, Libp2pEvents, Logger, TypedEventTarget, PeerId, PeerStore, Peer, PeerData, PeerQuery, PeerInfo, AbortOptions, ConsumePeerRecordOptions } from '@libp2p/interface'
+import type { ComponentLogger, Libp2pEvents, Logger, TypedEventTarget, PeerId, PeerStore, Peer, PeerData, PeerQuery, PeerInfo, AbortOptions, ConsumePeerRecordOptions, Metrics } from '@libp2p/interface'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { Datastore } from 'interface-datastore'
 
@@ -19,6 +19,7 @@ export interface PersistentPeerStoreComponents {
   datastore: Datastore
   events: TypedEventTarget<Libp2pEvents>
   logger: ComponentLogger
+  metrics?: Metrics
 }
 
 /**
@@ -75,29 +76,17 @@ class PersistentPeerStore implements PeerStore {
   readonly [Symbol.toStringTag] = '@libp2p/peer-store'
 
   async forEach (fn: (peer: Peer,) => void, query?: PeerQuery): Promise<void> {
-    const release = await this.store.lock.readLock(query)
-
-    try {
-      for await (const peer of this.store.all(query)) {
-        fn(peer)
-      }
-    } finally {
-      release()
+    for await (const peer of this.store.all(query)) {
+      fn(peer)
     }
   }
 
   async all (query?: PeerQuery): Promise<Peer[]> {
-    const release = await this.store.lock.readLock(query)
-
-    try {
-      return await all(this.store.all(query))
-    } finally {
-      release()
-    }
+    return all(this.store.all(query))
   }
 
   async delete (peerId: PeerId, options?: AbortOptions): Promise<void> {
-    const release = await this.store.lock.writeLock(options)
+    const release = await this.store.getReadLock(peerId, options)
 
     try {
       await this.store.delete(peerId, options)
@@ -107,23 +96,23 @@ class PersistentPeerStore implements PeerStore {
   }
 
   async has (peerId: PeerId, options?: AbortOptions): Promise<boolean> {
-    const release = await this.store.lock.readLock(options)
+    const release = await this.store.getReadLock(peerId, options)
 
     try {
       return await this.store.has(peerId, options)
     } finally {
       this.log.trace('has release read lock')
-      release()
+      release?.()
     }
   }
 
   async get (peerId: PeerId, options?: AbortOptions): Promise<Peer> {
-    const release = await this.store.lock.readLock(options)
+    const release = await this.store.getReadLock(peerId, options)
 
     try {
       return await this.store.load(peerId, options)
     } finally {
-      release()
+      release?.()
     }
   }
 
@@ -137,7 +126,7 @@ class PersistentPeerStore implements PeerStore {
   }
 
   async save (id: PeerId, data: PeerData, options?: AbortOptions): Promise<Peer> {
-    const release = await this.store.lock.writeLock(options)
+    const release = await this.store.getWriteLock(id, options)
 
     try {
       const result = await this.store.save(id, data, options)
@@ -146,12 +135,12 @@ class PersistentPeerStore implements PeerStore {
 
       return result.peer
     } finally {
-      release()
+      release?.()
     }
   }
 
   async patch (id: PeerId, data: PeerData, options?: AbortOptions): Promise<Peer> {
-    const release = await this.store.lock.writeLock(options)
+    const release = await this.store.getWriteLock(id, options)
 
     try {
       const result = await this.store.patch(id, data, options)
@@ -160,12 +149,12 @@ class PersistentPeerStore implements PeerStore {
 
       return result.peer
     } finally {
-      release()
+      release?.()
     }
   }
 
   async merge (id: PeerId, data: PeerData, options?: AbortOptions): Promise<Peer> {
-    const release = await this.store.lock.writeLock(options)
+    const release = await this.store.getWriteLock(id, options)
 
     try {
       const result = await this.store.merge(id, data, options)
@@ -174,7 +163,7 @@ class PersistentPeerStore implements PeerStore {
 
       return result.peer
     } finally {
-      release()
+      release?.()
     }
   }
 

packages/peer-store/src/store.ts

@@ -1,4 +1,5 @@
 import { NotFoundError } from '@libp2p/interface'
+import { PeerMap, trackedPeerMap } from '@libp2p/peer-collections'
 import { peerIdFromCID } from '@libp2p/peer-id'
 import mortice from 'mortice'
 import { base32 } from 'multiformats/bases/base32'
@@ -13,7 +14,7 @@ import type { AddressFilter, PersistentPeerStoreComponents, PersistentPeerStoreI
 import type { PeerUpdate as PeerUpdateExternal, PeerId, Peer, PeerData, PeerQuery, Logger } from '@libp2p/interface'
 import type { AbortOptions } from '@multiformats/multiaddr'
 import type { Datastore, Key, Query } from 'interface-datastore'
-import type { Mortice } from 'mortice'
+import type { Mortice, Release } from 'mortice'
 
 /**
  * Event detail emitted when peer data changes
@@ -53,10 +54,15 @@ function mapQuery (query: PeerQuery, maxAddressAge: number): Query {
   }
 }
 
+export interface Lock {
+  refs: number
+  lock: Mortice
+}
+
 export class PersistentStore {
   private readonly peerId: PeerId
   private readonly datastore: Datastore
-  public readonly lock: Mortice
+  private locks: PeerMap<Lock>
   private readonly addressFilter?: AddressFilter
   private readonly log: Logger
   private readonly maxAddressAge: number
@@ -67,14 +73,76 @@ export class PersistentStore {
     this.peerId = components.peerId
     this.datastore = components.datastore
     this.addressFilter = init.addressFilter
-    this.lock = mortice({
-      name: 'peer-store',
-      singleProcess: true
+    this.locks = trackedPeerMap({
+      name: 'libp2p_peer_store_locks',
+      metrics: components.metrics
     })
     this.maxAddressAge = init.maxAddressAge ?? MAX_ADDRESS_AGE
     this.maxPeerAge = init.maxPeerAge ?? MAX_PEER_AGE
   }
 
+  getLock (peerId: PeerId): Lock {
+    let lock = this.locks.get(peerId)
+
+    if (lock == null) {
+      lock = {
+        refs: 0,
+        lock: mortice({
+          name: peerId.toString(),
+          singleProcess: true
+        })
+      }
+
+      this.locks.set(peerId, lock)
+    }
+
+    lock.refs++
+
+    return lock
+  }
+
+  private maybeRemoveLock (peerId: PeerId, lock: Lock): void {
+    lock.refs--
+
+    if (lock.refs === 0) {
+      this.locks.delete(peerId)
+    }
+  }
+
+  async getReadLock (peerId: PeerId, options?: AbortOptions): Promise<Release> {
+    const lock = this.getLock(peerId)
+
+    try {
+      const release = await lock.lock.readLock(options)
+
+      return () => {
+        release()
+        this.maybeRemoveLock(peerId, lock)
+      }
+    } catch (err) {
+      this.maybeRemoveLock(peerId, lock)
+
+      throw err
+    }
+  }
+
+  async getWriteLock (peerId: PeerId, options?: AbortOptions): Promise<Release> {
+    const lock = this.getLock(peerId)
+
+    try {
+      const release = await lock.lock.writeLock(options)
+
+      return () => {
+        release()
+        this.maybeRemoveLock(peerId, lock)
+      }
+    } catch (err) {
+      this.maybeRemoveLock(peerId, lock)
+
+      throw err
+    }
+  }
+
   async has (peerId: PeerId, options?: AbortOptions): Promise<boolean> {
     try {
       await this.load(peerId, options)
@@ -179,7 +247,7 @@ export class PersistentStore {
 
       return {
         peerPB,
-        peer: bytesToPeer(peerId, buf, this.maxAddressAge)
+        peer: pbToPeer(peerId, peerPB, this.maxAddressAge)
       }
     } catch (err: any) {
       if (err.name !== 'NotFoundError') {
@@ -196,7 +264,7 @@ export class PersistentStore {
     await this.datastore.put(peerIdToDatastoreKey(peerId), buf, options)
 
     return {
-      peer: bytesToPeer(peerId, buf, this.maxAddressAge),
+      peer: pbToPeer(peerId, peer, this.maxAddressAge),
       previous: existingPeer?.peer,
       updated: existingPeer == null || !peerEquals(peer, existingPeer.peerPB)
     }

packages/peer-store/src/utils/bytes-to-peer.ts

@@ -1,22 +1,20 @@
 import { publicKeyFromProtobuf } from '@libp2p/crypto/keys'
 import { peerIdFromPublicKey } from '@libp2p/peer-id'
 import { multiaddr } from '@multiformats/multiaddr'
-import { base58btc } from 'multiformats/bases/base58'
-import * as Digest from 'multiformats/hashes/digest'
 import { Peer as PeerPB } from '../pb/peer.js'
 import type { PeerId, Peer, Tag } from '@libp2p/interface'
+import type { Digest } from 'multiformats/hashes/digest'
 
 function populatePublicKey (peerId: PeerId, protobuf: PeerPB): PeerId {
   if (peerId.publicKey != null || protobuf.publicKey == null) {
     return peerId
   }
 
-  let digest: any
+  let digest: Digest<18, number> | undefined
 
   if (peerId.type === 'RSA') {
     // avoid hashing public key
-    const multihash = base58btc.decode(`z${peerId}`)
-    digest = Digest.decode(multihash)
+    digest = peerId.toMultihash()
   }
 
   const publicKey = publicKeyFromProtobuf(protobuf.publicKey, digest)

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

@@ -252,8 +252,13 @@ function mapTag (key: string, tag: any): Tag {
     expiry = BigInt(Date.now() + Number(tag.ttl))
   }
 
-  return {
-    value: tag.value ?? 0,
-    expiry
+  const output: Tag = {
+    value: tag.value ?? 0
   }
+
+  if (expiry != null) {
+    output.expiry = expiry
+  }
+
+  return output
 }

packages/peer-store/test/index.spec.ts

@@ -558,4 +558,34 @@ describe('PersistentPeerStore', () => {
       ]
     })
   })
+
+  it('should allow concurrent merging', async () => {
+    const peerStore = persistentPeerStore(components)
+
+    expect(peerStore).to.have.nested.property('store.locks')
+      .that.has.property('size', 0)
+
+    const p1 = peerStore.merge(otherPeerId, {
+      multiaddrs: [
+        multiaddr('/ip4/123.123.123.123/tcp/1234')
+      ]
+    })
+
+    const p2 = peerStore.merge(otherPeerId, {
+      multiaddrs: [
+        multiaddr('/ip4/123.123.123.123/tcp/4567')
+      ]
+    })
+
+    expect(peerStore).to.have.nested.property('store.locks')
+      .that.has.property('size', 1)
+
+    await Promise.all([p1, p2])
+
+    const peer = await peerStore.get(otherPeerId)
+
+    expect(peer.addresses).to.have.lengthOf(2)
+    expect(peerStore).to.have.nested.property('store.locks')
+      .that.has.property('size', 0)
+  })
 })