fix: remove provider lock (#3169)

This creates resource contention and is not necessary.

Author: achingbrain

packages/kad-dht/package.json

@@ -68,7 +68,6 @@
     "it-protobuf-stream": "^2.0.2",
     "it-pushable": "^3.2.3",
     "it-take": "^3.0.8",
-    "mortice": "^3.0.6",
     "multiformats": "^13.3.6",
     "p-defer": "^4.0.1",
     "p-event": "^6.0.1",

packages/kad-dht/src/kad-dht.ts

@@ -1,6 +1,5 @@
 import { NotFoundError, TypedEventEmitter, contentRoutingSymbol, peerDiscoverySymbol, peerRoutingSymbol, serviceCapabilities, serviceDependencies, setMaxListeners, start, stop } from '@libp2p/interface'
 import drain from 'it-drain'
-import createMortice from 'mortice'
 import pDefer from 'p-defer'
 import { ALPHA, ON_PEER_CONNECT_TIMEOUT, PROTOCOL } from './constants.js'
 import { ContentFetching } from './content-fetching/index.js'
@@ -175,13 +174,10 @@ export class KadDHT extends TypedEventEmitter<PeerDiscoveryEvents> implements Ka
     this.peerInfoMapper = init.peerInfoMapper ?? removePrivateAddressesMapper
     this.onPeerConnectTimeout = init.onPeerConnectTimeout ?? ON_PEER_CONNECT_TIMEOUT
 
-    const providerLock = createMortice()
-
     this.providers = new Providers(components, {
       ...init.providers,
       logPrefix,
-      datastorePrefix,
-      lock: providerLock
+      datastorePrefix
     })
 
     this.validators = {
@@ -292,7 +288,6 @@ export class KadDHT extends TypedEventEmitter<PeerDiscoveryEvents> implements Ka
       metricsPrefix,
       datastorePrefix,
       contentRouting: this.contentRouting,
-      lock: providerLock,
       operationMetrics
     })
 

packages/kad-dht/src/providers.ts

@@ -3,13 +3,11 @@ import * as varint from 'uint8-varint'
 import { parseProviderKey, readProviderTime, toProviderKey } from './utils.js'
 import type { AbortOptions, ComponentLogger, Logger, Metrics, PeerId } from '@libp2p/interface'
 import type { Datastore } from 'interface-datastore'
-import type { Mortice } from 'mortice'
 import type { CID } from 'multiformats'
 
 export interface ProvidersInit {
   logPrefix: string
   datastorePrefix: string
-  lock: Mortice
 }
 
 export interface ProvidersComponents {
@@ -29,59 +27,39 @@ export class Providers {
   private readonly log: Logger
   private readonly datastore: Datastore
   private readonly datastorePrefix: string
-  private readonly lock: Mortice
 
   constructor (components: ProvidersComponents, init: ProvidersInit) {
     this.log = components.logger.forComponent(`${init.logPrefix}:providers`)
     this.datastorePrefix = `${init.datastorePrefix}/provider`
     this.datastore = components.datastore
-    this.lock = init.lock
   }
 
   /**
    * Add a new provider for the given CID
    */
   async addProvider (cid: CID, provider: PeerId, options?: AbortOptions): Promise<void> {
-    const release = await this.lock.readLock(options)
-
-    try {
-      this.log.trace('%p provides %s', provider, cid)
-      await this.writeProviderEntry(cid, provider, options)
-    } finally {
-      release()
-    }
+    this.log.trace('%p provides %s', provider, cid)
+    await this.writeProviderEntry(cid, provider, options)
   }
 
   /**
    * Remove a provider for the given CID
    */
   async removeProvider (cid: CID, provider: PeerId, options?: AbortOptions): Promise<void> {
-    const release = await this.lock.writeLock(options)
-
-    try {
-      const key = toProviderKey(this.datastorePrefix, cid, provider)
-      this.log.trace('%p no longer provides %s', provider, cid)
-      await this.datastore.delete(key)
-    } finally {
-      release()
-    }
+    const key = toProviderKey(this.datastorePrefix, cid, provider)
+    this.log.trace('%p no longer provides %s', provider, cid)
+    await this.datastore.delete(key, options)
   }
 
   /**
    * Get a list of providers for the given CID
    */
   async getProviders (cid: CID, options?: AbortOptions): Promise<PeerId[]> {
-    const release = await this.lock.readLock(options)
-
-    try {
-      this.log.trace('get providers for %c', cid)
-      const provs = await this.loadProviders(cid, options)
-      this.log.trace('got %d providers for %c', provs.size, cid)
+    this.log.trace('get providers for %c', cid)
+    const provs = await this.loadProviders(cid, options)
+    this.log.trace('got %d providers for %c', provs.size, cid)
 
-      return [...provs.keys()]
-    } finally {
-      release()
-    }
+    return [...provs.keys()]
   }
 
   /**

packages/kad-dht/src/reprovider.ts

@@ -10,7 +10,6 @@ import type { AbortOptions, ComponentLogger, Logger, Metrics, PeerId } from '@li
 import type { AddressManager } from '@libp2p/interface-internal'
 import type { AdaptiveTimeoutInit } from '@libp2p/utils/adaptive-timeout'
 import type { Datastore } from 'interface-datastore'
-import type { Mortice } from 'mortice'
 import type { CID } from 'multiformats/cid'
 
 export interface ReproviderComponents {
@@ -26,7 +25,6 @@ export interface ReproviderInit {
   metricsPrefix: string
   datastorePrefix: string
   contentRouting: ContentRouting
-  lock: Mortice
   operationMetrics: OperationMetrics
   concurrency?: number
   maxQueueSize?: number
@@ -60,7 +58,6 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
   private readonly addressManager: AddressManager
   private readonly validity: number
   private readonly interval: number
-  private readonly lock: Mortice
   private readonly peerId: PeerId
 
   constructor (components: ReproviderComponents, init: ReproviderInit) {
@@ -86,7 +83,6 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
     this.validity = init.validity ?? PROVIDERS_VALIDITY
     this.interval = init.interval ?? REPROVIDE_INTERVAL
     this.contentRouting = init.contentRouting
-    this.lock = init.lock
     this.running = false
 
     this.reprovide = timeOperationMethod(this.reprovide.bind(this), init.operationMetrics, 'PROVIDE')
@@ -123,15 +119,13 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
    * them if the provider is us and the expiry is within the reprovide window.
    */
   private async cleanUp (options?: AbortOptions): Promise<void> {
-    const release = await this.lock.writeLock(options)
-
     try {
       this.safeDispatchEvent('reprovide:start')
 
       // Get all provider entries from the datastore
       for await (const entry of this.datastore.query({
         prefix: this.datastorePrefix
-      })) {
+      }, options)) {
         try {
           // Add a delete to the batch for each expired entry
           const { cid, peerId } = parseProviderKey(entry.key)
@@ -144,7 +138,7 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
 
           // delete the record if it has expired
           if (expired) {
-            await this.datastore.delete(entry.key)
+            await this.datastore.delete(entry.key, options)
           }
 
           // if the provider is us and we are within the reprovide threshold,
@@ -162,7 +156,6 @@ export class Reprovider extends TypedEventEmitter<ReprovideEvents> {
 
       this.log('reprovide/cleanup successful')
     } finally {
-      release()
       this.safeDispatchEvent('reprovide:end')
 
       if (this.running) {

packages/kad-dht/src/rpc/index.ts

@@ -15,7 +15,7 @@ import type { GetProvidersHandlerComponents } from './handlers/get-providers.js'
 import type { GetValueHandlerComponents } from './handlers/get-value.js'
 import type { PutValueHandlerComponents } from './handlers/put-value.js'
 import type { RoutingTable } from '../routing-table/index.js'
-import type { CounterGroup, Logger, Metrics, PeerId, IncomingStreamData } from '@libp2p/interface'
+import type { CounterGroup, Logger, Metrics, PeerId, IncomingStreamData, MetricGroup } from '@libp2p/interface'
 
 export interface DHTMessageHandler {
   handle(peerId: PeerId, msg: Message): Promise<Message | undefined>
@@ -43,14 +43,16 @@ export class RPC {
   private readonly metrics: {
     operations?: CounterGroup
     errors?: CounterGroup
+    rpcTime?: MetricGroup
   }
 
   private readonly incomingMessageTimeout: number
 
   constructor (components: RPCComponents, init: RPCInit) {
     this.metrics = {
       operations: components.metrics?.registerCounterGroup(`${init.metricsPrefix}_inbound_rpc_requests_total`),
-      errors: components.metrics?.registerCounterGroup(`${init.metricsPrefix}_inbound_rpc_errors_total`)
+      errors: components.metrics?.registerCounterGroup(`${init.metricsPrefix}_inbound_rpc_errors_total`),
+      rpcTime: components.metrics?.registerMetricGroup(`${init.metricsPrefix}_inbound_rpc_time_seconds`, { label: 'operation' })
     }
 
     this.log = components.logger.forComponent(`${init.logPrefix}:rpc`)
@@ -114,15 +116,30 @@ export class RPC {
             signal
           })
 
-          // handle the message
-          this.log('incoming %s from %p', message.type, connection.remotePeer)
-          const res = await this.handleMessage(connection.remotePeer, message)
-
-          // Not all handlers will return a response
-          if (res != null) {
-            await messages.write(res, {
-              signal
-            })
+          const stopSuccessTimer = this.metrics?.rpcTime?.timer(message.type.toString())
+          const stopErrorTimer = this.metrics?.rpcTime?.timer(message.type.toString())
+          let errored = false
+
+          try {
+            // handle the message
+            this.log('incoming %s from %p', message.type, connection.remotePeer)
+            const res = await this.handleMessage(connection.remotePeer, message)
+
+            // Not all handlers will return a response
+            if (res != null) {
+              await messages.write(res, {
+                signal
+              })
+            }
+          } catch (err) {
+            errored = true
+            stopErrorTimer?.()
+
+            throw err
+          } finally {
+            if (!errored) {
+              stopSuccessTimer?.()
+            }
           }
 
           // we have received a message so reset the timeout controller to

packages/kad-dht/test/providers.spec.ts

@@ -4,7 +4,6 @@ import { defaultLogger } from '@libp2p/logger'
 import { peerIdFromString } from '@libp2p/peer-id'
 import { expect } from 'aegir/chai'
 import { MemoryDatastore } from 'datastore-core/memory'
-import createMortice from 'mortice'
 import { CID } from 'multiformats/cid'
 import { Providers } from '../src/providers.js'
 import { createPeerIdsWithPrivateKey } from './utils/create-peer-id.js'
@@ -24,8 +23,7 @@ describe('providers', () => {
       logger: defaultLogger()
     }, {
       logPrefix: '',
-      datastorePrefix: '/dht',
-      lock: createMortice()
+      datastorePrefix: '/dht'
     })
 
     const cid = CID.parse('QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n')
@@ -46,8 +44,7 @@ describe('providers', () => {
       logger: defaultLogger()
     }, {
       logPrefix: '',
-      datastorePrefix: '/dht',
-      lock: createMortice()
+      datastorePrefix: '/dht'
     })
 
     const cid = CID.parse('QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n')
@@ -72,8 +69,7 @@ describe('providers', () => {
       logger: defaultLogger()
     }, {
       logPrefix: '',
-      datastorePrefix: '/dht',
-      lock: createMortice()
+      datastorePrefix: '/dht'
     })
 
     const cid = CID.parse('QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n')
@@ -101,8 +97,7 @@ describe('providers', () => {
       logger: defaultLogger()
     }, {
       logPrefix: '',
-      datastorePrefix: '/dht',
-      lock: createMortice()
+      datastorePrefix: '/dht'
     })
 
     const cid = CID.parse('QmZ8eiDPqQqWR17EPxiwCDgrKPVhCHLcyn6xSCNpFAdAZb')

packages/kad-dht/test/reprovider.spec.ts

@@ -3,7 +3,6 @@ import { defaultLogger } from '@libp2p/logger'
 import { expect } from 'aegir/chai'
 import { MemoryDatastore } from 'datastore-core'
 import delay from 'delay'
-import createMortice from 'mortice'
 import { CID } from 'multiformats/cid'
 import { pEvent } from 'p-event'
 import { stubInterface } from 'sinon-ts'
@@ -45,22 +44,19 @@ describe('reprovider', () => {
     contentRouting = stubInterface()
     contentRouting.provide.resolves([])
 
-    const lock = createMortice()
     const logPrefix = 'libp2p'
     const datastorePrefix = '/dht'
     const metricsPrefix = ''
 
     providers = new Providers(components, {
       logPrefix,
-      datastorePrefix,
-      lock
+      datastorePrefix
     })
 
     reprovider = new Reprovider(components, {
       logPrefix,
       datastorePrefix,
       metricsPrefix,
-      lock,
       contentRouting,
       threshold: 100,
       validity: 200,

packages/kad-dht/test/rpc/handlers/add-provider.spec.ts

@@ -6,7 +6,6 @@ import { persistentPeerStore } from '@libp2p/peer-store'
 import { multiaddr } from '@multiformats/multiaddr'
 import { expect } from 'aegir/chai'
 import { MemoryDatastore } from 'datastore-core'
-import createMortice from 'mortice'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { MessageType } from '../../../src/message/dht.js'
 import { Providers } from '../../../src/providers.js'
@@ -47,8 +46,7 @@ describe('rpc - handlers - AddProvider', () => {
       logger: defaultLogger()
     }, {
       logPrefix: '',
-      datastorePrefix: '/dht',
-      lock: createMortice()
+      datastorePrefix: '/dht'
     })
 
     handler = new AddProviderHandler({