fix: prefix records with key, remove disjoint queries (#239)

1. Prefix dht records with `/dht/record` and dht providers with `/dht/provider` in the datastore to prevent accidentally overwriting the keys in other parts of the codebase
2. Queries are no longer disjoint as it's not in the spec (though we keep a map of peers seen per query to prevent loops)
3. When looking up providers, do not query the datastore for the bare CID thinking we'll find ourselves as a provider as that's not how the provider record keys are stored
4. Add typedefs for query event types so they are in the generated `index.d.ts`

Author: achingbrain

README.md

@@ -29,7 +29,7 @@
   - [Peer Routing](#peer-routing)
   - [Content Routing](#content-routing)
   - [Peer Discovery](#peer-discovery)
-  - [Implementation Summary](#implementation-summary)
+- [Spec](#spec)
 - [Contribute](#contribute)
 - [License](#license)
 ## Install
@@ -84,9 +84,9 @@ Note that you may want to supply your own peer discovery function and datastore
 
 [![](https://raw.githubusercontent.com/libp2p/js-libp2p-interfaces/master/src/peer-discovery/img/badge.png)](https://github.com/libp2p/js-libp2p-interfaces/tree/master/packages/interfaces/src/peer-discovery)
 
-### Implementation Summary
+## Spec
 
-A [summary](docs/IMPL_SUMMARY.MD) of the algorithms and API for this implementation of Kademlia.
+js-libp2p-kad-dht follows the [libp2p/kad-dht spec](https://github.com/libp2p/specs/tree/master/kad-dht) and implements the algorithms described in the [IPFS DHT documentation](https://docs.ipfs.io/concepts/dht/).
 
 ## Contribute
 

docs/IMPL_SUMMARY.MD

@@ -15,7 +15,9 @@ exports.MAX_RECORD_AGE = 36 * hour
 
 exports.PROTOCOL_DHT = '/kad/1.0.0'
 
-exports.PROVIDERS_KEY_PREFIX = '/providers/'
+exports.RECORD_KEY_PREFIX = '/dht/record/'
+
+exports.PROVIDER_KEY_PREFIX = '/dht/provider/'
 
 exports.PROVIDERS_LRU_CACHE_SIZE = 256
 

src/constants.js

@@ -27,7 +27,7 @@ class ContentFetching {
   /**
    * @param {object} params
    * @param {import('peer-id')} params.peerId
-   * @param {import('interface-datastore').Datastore} params.datastore
+   * @param {import('interface-datastore').Datastore} params.records
    * @param {import('libp2p-interfaces/src/types').DhtValidators} params.validators
    * @param {import('libp2p-interfaces/src/types').DhtSelectors} params.selectors
    * @param {import('../peer-routing').PeerRouting} params.peerRouting
@@ -36,10 +36,10 @@ class ContentFetching {
    * @param {import('../network').Network} params.network
    * @param {boolean} params.lan
    */
-  constructor ({ peerId, datastore, validators, selectors, peerRouting, queryManager, routingTable, network, lan }) {
+  constructor ({ peerId, records, validators, selectors, peerRouting, queryManager, routingTable, network, lan }) {
     this._log = utils.logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:content-fetching`)
     this._peerId = peerId
-    this._datastore = datastore
+    this._records = records
     this._validators = validators
     this._selectors = selectors
     this._peerRouting = peerRouting
@@ -53,7 +53,7 @@ class ContentFetching {
    * @param {Uint8Array} rec
    */
   async putLocal (key, rec) { // eslint-disable-line require-await
-    return this._datastore.put(utils.bufferToKey(key), rec)
+    return this._records.put(utils.bufferToKey(key), rec)
   }
 
   /**
@@ -68,7 +68,7 @@ class ContentFetching {
     const dsKey = utils.bufferToKey(key)
 
     this._log(`fetching record for key ${dsKey}`)
-    const raw = await this._datastore.get(dsKey)
+    const raw = await this._records.get(dsKey)
     this._log(`found ${dsKey} in local datastore`)
 
     const rec = Record.deserialize(raw)
@@ -103,7 +103,7 @@ class ContentFetching {
         try {
           const dsKey = utils.bufferToKey(key)
           this._log(`Storing corrected record for key ${dsKey}`)
-          await this._datastore.put(dsKey, fixupRec)
+          await this._records.put(dsKey, fixupRec)
         } catch (/** @type {any} */ err) {
           this._log.error('Failed error correcting self', err)
         }
@@ -149,7 +149,7 @@ class ContentFetching {
     // store the record locally
     const dsKey = utils.bufferToKey(key)
     this._log(`storing record for key ${dsKey}`)
-    await this._datastore.put(dsKey, record)
+    await this._records.put(dsKey, record)
 
     // put record to the closest peers
     yield * pipe(

src/content-fetching/index.js

@@ -2,10 +2,10 @@
 
 const { EventEmitter } = require('events')
 const PeerId = require('peer-id')
-const { toString: uint8ArrayToString } = require('uint8arrays/to-string')
 const utils = require('./utils')
 const errCode = require('err-code')
 const merge = require('it-merge')
+const { queryErrorEvent } = require('./query/events')
 
 const log = utils.logger('libp2p:kad-dht')
 
@@ -57,7 +57,6 @@ class DualKadDHT extends EventEmitter {
     this._wan = wan
     this._lan = lan
     this._libp2p = libp2p
-    this._datastore = libp2p.datastore || this._wan._datastore
 
     // handle peers being discovered during processing of DHT messages
     this._wan.on('peer', (peerData) => {
@@ -159,6 +158,7 @@ class DualKadDHT extends EventEmitter {
    */
   async * get (key, options = {}) { // eslint-disable-line require-await
     let queriedPeers = false
+    let foundValue = false
 
     for await (const event of merge(
       this._lan.get(key, options),
@@ -172,6 +172,10 @@ class DualKadDHT extends EventEmitter {
 
       if (event.name === 'VALUE') {
         queriedPeers = true
+
+        if (event.value != null) {
+          foundValue = true
+        }
       }
 
       if (event.name === 'SENDING_QUERY') {
@@ -182,24 +186,12 @@ class DualKadDHT extends EventEmitter {
     if (!queriedPeers) {
       throw errCode(new Error('No peers found in routing table!'), 'ERR_NO_PEERS_IN_ROUTING_TABLE')
     }
-  }
 
-  /**
-   * Remove the given key from the local datastore
-   *
-   * @param {Uint8Array} key
-   */
-  async removeLocal (key) {
-    log(`removeLocal: ${uint8ArrayToString(key, 'base32')}`)
-    const dsKey = utils.bufferToKey(key)
-
-    try {
-      await this._datastore.delete(dsKey)
-    } catch (/** @type {any} */ err) {
-      if (err.code === 'ERR_NOT_FOUND') {
-        return undefined
-      }
-      throw err
+    if (!foundValue) {
+      yield queryErrorEvent({
+        from: this._libp2p.peerId,
+        error: errCode(new Error('Not found'), 'ERR_NOT_FOUND')
+      })
     }
   }
 

src/dual-kad-dht.js

@@ -6,6 +6,15 @@ const { DualKadDHT } = require('./dual-kad-dht')
 /**
  * @typedef {import('./types').DHT} DHT
  * @typedef {import('./kad-dht').KadDHTOps} KadDHTOps
+ * @typedef {import('./types').QueryEvent} QueryEvent
+ * @typedef {import('./types').SendingQueryEvent} SendingQueryEvent
+ * @typedef {import('./types').PeerResponseEvent} PeerResponseEvent
+ * @typedef {import('./types').FinalPeerEvent} FinalPeerEvent
+ * @typedef {import('./types').QueryErrorEvent} QueryErrorEvent
+ * @typedef {import('./types').ProviderEvent} ProviderEvent
+ * @typedef {import('./types').ValueEvent} ValueEvent
+ * @typedef {import('./types').AddingPeerEvent} AddingPeerEvent
+ * @typedef {import('./types').DialingPeerEvent} DialingPeerEvent
  */
 
 module.exports = {

src/index.js

@@ -9,7 +9,8 @@ const { RoutingTableRefresh } = require('./routing-table/refresh')
 const utils = require('./utils')
 const {
   K,
-  QUERY_SELF_INTERVAL
+  QUERY_SELF_INTERVAL,
+  RECORD_KEY_PREFIX
 } = require('./constants')
 const { Network } = require('./network')
 const { ContentFetching } = require('./content-fetching')
@@ -24,6 +25,8 @@ const {
   removePrivateAddresses,
   removePublicAddresses
 } = require('./utils')
+const { KeyTransformDatastore } = require('datastore-core')
+const { Key } = require('interface-datastore/key')
 
 /**
  * @typedef {import('libp2p')} Libp2p
@@ -58,6 +61,40 @@ const {
  * @property {PeerData[]} bootstrapPeers
  */
 
+class PrefixTransform {
+  /**
+   *
+   * @param {string} prefix - : ;
+   */
+  constructor (prefix) {
+    this._prefix = prefix
+
+    if (this._prefix.startsWith('/')) {
+      this._prefix = this._prefix.substring(1)
+    }
+  }
+
+  /**
+   * @param {Key} key
+   */
+  convert (key) {
+    return new Key(`/${this._prefix}${key}`)
+  }
+
+  /**
+   * @param {Key} key
+   */
+  invert (key) {
+    const namespaces = key.namespaces()
+
+    if (namespaces[0] === this._prefix) {
+      namespaces.shift()
+    }
+
+    return Key.withNamespaces(namespaces)
+  }
+}
+
 /**
  * A DHT implementation modelled after Kademlia with S/Kademlia modifications.
  * Original implementation in go: https://github.com/libp2p/go-libp2p-kad-dht.
@@ -127,19 +164,17 @@ class KadDHT extends EventEmitter {
       lan
     })
 
-    /**
-     * Reference to the datastore, uses an in-memory store if none given.
-     *
-     * @type {Datastore}
-     */
-    this._datastore = libp2p.datastore || new MemoryDatastore()
+    const datastore = libp2p.datastore || new MemoryDatastore()
+    const records = new KeyTransformDatastore(datastore, new PrefixTransform(RECORD_KEY_PREFIX))
 
     /**
      * Provider management
      *
      * @type {Providers}
      */
-    this._providers = new Providers(this._datastore)
+    this._providers = new Providers({
+      providers: datastore
+    })
 
     /**
      * @type {boolean}
@@ -185,7 +220,7 @@ class KadDHT extends EventEmitter {
     })
     this._contentFetching = new ContentFetching({
       peerId: libp2p.peerId,
-      datastore: this._datastore,
+      records,
       validators: this._validators,
       selectors: this._selectors,
       peerRouting: this._peerRouting,
@@ -216,7 +251,7 @@ class KadDHT extends EventEmitter {
       peerStore: libp2p.peerStore,
       addressable: libp2p,
       peerRouting: this._peerRouting,
-      datastore: this._datastore,
+      records,
       validators: this._validators,
       lan
     })