feat: add reprovide (#2785)

Adds a reprovide queue to kad-dht to republish provider records two
hours before they would expire.

Adds a function to the content routing interface to cancel the
reprovide.

Author: achingbrain

packages/integration-tests/test/fixtures/utils.ts

@@ -166,6 +166,10 @@ export class MockContentRouting implements ContentRouting {
     MockContentRouting.providers.set(cid.toString(), providers)
   }
 
+  async cancelReprovide (): Promise<void> {
+
+  }
+
   async * findProviders (cid: CID<unknown, number, number, Version>, options?: AbortOptions | undefined): AsyncGenerator<PeerInfo, void, undefined> {
     yield * MockContentRouting.providers.get(cid.toString()) ?? []
   }

packages/interface/src/content-routing/index.ts

@@ -1,4 +1,4 @@
-import type { RoutingOptions } from '../index.js'
+import type { AbortOptions, RoutingOptions } from '../index.js'
 import type { PeerInfo } from '../peer-info/index.js'
 import type { CID } from 'multiformats/cid'
 
@@ -45,6 +45,13 @@ export interface ContentRouting {
    */
   provide(cid: CID, options?: RoutingOptions): Promise<void>
 
+  /**
+   * If network peers need to be periodically reminded that the caller can
+   * provide content corresponding to the passed CID, call this function to no
+   * longer remind them.
+   */
+  cancelReprovide (key: CID, options?: AbortOptions): Promise<void>
+
   /**
    * Find the providers of the passed CID.
    *

packages/kad-dht/README.md

@@ -102,6 +102,42 @@ const peerInfo = await node.peerRouting.findPeer(peerId)
 console.info(peerInfo) // peer id, multiaddrs
 ```
 
+## Example - Connecting to both a LAN-only DHT and the IPFS Amino DHT
+
+When using multiple DHTs, you should specify distinct datastore, metrics and
+log prefixes to ensure that data is kept separate for each instance.
+
+```TypeScript
+import { kadDHT, removePublicAddressesMapper, removePrivateAddressesMapper } from '@libp2p/kad-dht'
+import { createLibp2p } from 'libp2p'
+import { peerIdFromString } from '@libp2p/peer-id'
+
+const node = await createLibp2p({
+  services: {
+    lanDHT: kadDHT({
+      protocol: '/ipfs/lan/kad/1.0.0',
+      peerInfoMapper: removePublicAddressesMapper,
+      clientMode: false,
+      logPrefix: 'libp2p:dht-lan',
+      datastorePrefix: '/dht-lan',
+      metricsPrefix: 'libp2p_dht_lan'
+    }),
+    aminoDHT: kadDHT({
+      protocol: '/ipfs/kad/1.0.0',
+      peerInfoMapper: removePrivateAddressesMapper,
+      logPrefix: 'libp2p:dht-amino',
+      datastorePrefix: '/dht-amino',
+      metricsPrefix: 'libp2p_dht_amino'
+    })
+  }
+})
+
+const peerId = peerIdFromString('QmFoo')
+const peerInfo = await node.peerRouting.findPeer(peerId)
+
+console.info(peerInfo) // peer id, multiaddrs
+```
+
 # Install
 
 ```console

packages/kad-dht/package.json

@@ -66,7 +66,6 @@
     "@libp2p/utils": "^6.1.2",
     "@multiformats/multiaddr": "^12.2.3",
     "any-signal": "^4.1.1",
-    "hashlru": "^2.3.0",
     "interface-datastore": "^8.3.0",
     "it-drain": "^3.0.7",
     "it-length": "^3.0.6",
@@ -77,10 +76,10 @@
     "it-pipe": "^3.0.1",
     "it-protobuf-stream": "^1.1.3",
     "it-take": "^3.0.5",
+    "mortice": "^3.0.4",
     "multiformats": "^13.1.0",
     "p-defer": "^4.0.1",
     "p-event": "^6.0.1",
-    "p-queue": "^8.0.1",
     "progress-events": "^1.0.0",
     "protons-runtime": "^5.4.0",
     "race-signal": "^1.0.2",
@@ -98,7 +97,6 @@
     "@types/which": "^3.0.3",
     "aegir": "^44.0.1",
     "datastore-core": "^10.0.0",
-    "datastore-level": "^11.0.0",
     "delay": "^6.0.0",
     "execa": "^9.1.0",
     "it-all": "^3.0.6",

packages/kad-dht/src/constants.ts

@@ -15,15 +15,21 @@ export const MAX_RECORD_AGE = 36 * hour
 
 export const PROTOCOL = '/ipfs/kad/1.0.0'
 
-export const RECORD_KEY_PREFIX = '/dht/record'
+export const PROVIDERS_VALIDITY = 24 * hour
 
-export const PROVIDER_KEY_PREFIX = '/dht/provider'
+export const PROVIDERS_CLEANUP_INTERVAL = hour
 
-export const PROVIDERS_LRU_CACHE_SIZE = 256
+// Re-run the provide operation when the expiry of our provider records is within this amount
+export const REPROVIDE_THRESHOLD = 2 * hour
 
-export const PROVIDERS_VALIDITY = 24 * hour
+// How many reprovide operations to run at once
+export const REPROVIDE_CONCURRENCY = 10
 
-export const PROVIDERS_CLEANUP_INTERVAL = hour
+// How long to let the reprovide queue grow before we wait for capacity
+export const REPROVIDE_MAX_QUEUE_SIZE = 16_384
+
+// How often to check if records need reproviding
+export const REPROVIDE_INTERVAL = hour
 
 export const READ_MESSAGE_TIMEOUT = 10 * second
 

packages/kad-dht/src/content-fetching/index.ts

@@ -41,12 +41,14 @@ export class ContentFetching {
   private readonly peerRouting: PeerRouting
   private readonly queryManager: QueryManager
   private readonly network: Network
+  private readonly datastorePrefix: string
 
   constructor (components: KadDHTComponents, init: ContentFetchingInit) {
     const { validators, selectors, peerRouting, queryManager, network, logPrefix } = init
 
     this.components = components
     this.log = components.logger.forComponent(`${logPrefix}:content-fetching`)
+    this.datastorePrefix = `/${init.logPrefix.replaceAll(':', '/')}/record`
     this.validators = validators
     this.selectors = selectors
     this.peerRouting = peerRouting
@@ -61,7 +63,7 @@ export class ContentFetching {
   async getLocal (key: Uint8Array): Promise<Libp2pRecord> {
     this.log('getLocal %b', key)
 
-    const dsKey = bufferToRecordKey(key)
+    const dsKey = bufferToRecordKey(this.datastorePrefix, key)
 
     this.log('fetching record for key %k', dsKey)
 
@@ -92,7 +94,7 @@ export class ContentFetching {
       // correct ourself
       if (this.components.peerId.equals(from)) {
         try {
-          const dsKey = bufferToRecordKey(key)
+          const dsKey = bufferToRecordKey(this.datastorePrefix, key)
           this.log(`Storing corrected record for key ${dsKey.toString()}`)
           await this.components.datastore.put(dsKey, fixupRec.subarray())
         } catch (err: any) {
@@ -136,7 +138,7 @@ export class ContentFetching {
     const record = createPutRecord(key, value)
 
     // store the record locally
-    const dsKey = bufferToRecordKey(key)
+    const dsKey = bufferToRecordKey(this.datastorePrefix, key)
     this.log(`storing record for key ${dsKey.toString()}`)
     await this.components.datastore.put(dsKey, record.subarray())
 

packages/kad-dht/src/index.ts

@@ -78,12 +78,47 @@
  *
  * console.info(peerInfo) // peer id, multiaddrs
  * ```
+ *
+ * @example Connecting to both a LAN-only DHT and the IPFS Amino DHT
+ *
+ * When using multiple DHTs, you should specify distinct datastore, metrics and
+ * log prefixes to ensure that data is kept separate for each instance.
+ *
+ * ```TypeScript
+ * import { kadDHT, removePublicAddressesMapper, removePrivateAddressesMapper } from '@libp2p/kad-dht'
+ * import { createLibp2p } from 'libp2p'
+ * import { peerIdFromString } from '@libp2p/peer-id'
+ *
+ * const node = await createLibp2p({
+ *   services: {
+ *     lanDHT: kadDHT({
+ *       protocol: '/ipfs/lan/kad/1.0.0',
+ *       peerInfoMapper: removePublicAddressesMapper,
+ *       clientMode: false,
+ *       logPrefix: 'libp2p:dht-lan',
+ *       datastorePrefix: '/dht-lan',
+ *       metricsPrefix: 'libp2p_dht_lan'
+ *     }),
+ *     aminoDHT: kadDHT({
+ *       protocol: '/ipfs/kad/1.0.0',
+ *       peerInfoMapper: removePrivateAddressesMapper,
+ *       logPrefix: 'libp2p:dht-amino',
+ *       datastorePrefix: '/dht-amino',
+ *       metricsPrefix: 'libp2p_dht_amino'
+ *     })
+ *   }
+ * })
+ *
+ * const peerId = peerIdFromString('QmFoo')
+ * const peerInfo = await node.peerRouting.findPeer(peerId)
+ *
+ * console.info(peerInfo) // peer id, multiaddrs
+ * ```
  */
 
 import { KadDHT as KadDHTClass } from './kad-dht.js'
 import { MessageType } from './message/dht.js'
 import { removePrivateAddressesMapper, removePublicAddressesMapper, passthroughMapper } from './utils.js'
-import type { ProvidersInit } from './providers.js'
 import type { Libp2pEvents, ComponentLogger, TypedEventTarget, Metrics, PeerId, PeerInfo, PeerStore, RoutingOptions, PrivateKey } from '@libp2p/interface'
 import type { AddressManager, ConnectionManager, Registrar } from '@libp2p/interface-internal'
 import type { AdaptiveTimeoutInit } from '@libp2p/utils/src/adaptive-timeout.js'
@@ -251,6 +286,12 @@ export interface KadDHT {
    */
   provide(key: CID, options?: RoutingOptions): AsyncIterable<QueryEvent>
 
+  /**
+   * Provider records must be re-published every 24 hours - pass a previously
+   * provided CID here to not re-publish a record for it any more
+   */
+  cancelReprovide(key: CID): Promise<void>
+
   /**
    * Store the passed value under the passed key on the DHT
    */
@@ -299,7 +340,61 @@ export type Selectors = Record<string, SelectFn>
  */
 export type Validators = Record<string, ValidateFn>
 
-export type { ProvidersInit }
+export interface ProvidersInit {
+  /**
+   * @default 256
+   */
+  cacheSize?: number
+  /**
+   * How often invalid records are cleaned. (in seconds)
+   *
+   * @default 5400
+   */
+  cleanupInterval?: number
+  /**
+   * How long is a provider valid for. (in seconds)
+   *
+   * @default 86400
+   */
+  provideValidity?: number
+}
+
+export interface ReProvideInit {
+  /**
+   * How many re-provide operations to run simultaneously
+   *
+   * @default 10
+   */
+  concurrency?: number
+
+  /**
+   * How long to let the re-provide queue grow
+   *
+   * @default 16384
+   */
+  maxQueueSize?: number
+
+  /**
+   * How long before the record expiry to re-provide in ms
+   *
+   * @default 7200000
+   */
+  threshold?: number
+
+  /**
+   * How often to check which records need reproviding in ms
+   *
+   * @default 3600000
+   */
+  interval?: number
+
+  /**
+   * How long provider records are valid for in ms
+   *
+   * @default 86400000
+   */
+  validity?: number
+}
 
 export interface KadDHTInit {
   /**
@@ -396,6 +491,20 @@ export interface KadDHTInit {
    */
   logPrefix?: string
 
+  /**
+   * The datastore prefix to use
+   *
+   * @default "/dht"
+   */
+  datastorePrefix?: string
+
+  /**
+   * The metrics prefix to use
+   *
+   * @default "libp2p_kad_dht"
+   */
+  metricsPrefix?: string
+
   /**
    * Settings for how long to wait in ms when pinging DHT peers to decide if
    * they should be evicted from the routing table or not.
@@ -459,6 +568,11 @@ export interface KadDHTInit {
    */
   providers?: ProvidersInit
 
+  /**
+   * Initialization options for the Reprovider component
+   */
+  reprovide?: ReProvideInit
+
   /**
    * For every incoming and outgoing PeerInfo, override address configuration
    * with this filter.