fix: do not abort dht operation on error responses (#3001)

Response streams from DHT operations contain lots of message types, one of which is 'Error', but they are emitted for all sorts of reasons, failing to dial peers, etc and valid responses can arrive after several errors.

The change here is to ignore error messages sent by the remote node as they do not indicate that the request has failed.

Fixes #2991

Author: achingbrain

packages/ipfs-http-client/src/dht/find-peer.js

@@ -5,6 +5,7 @@ const CID = require('cids')
 const multiaddr = require('multiaddr')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
+const { FinalPeer } = require('./response-types')
 
 module.exports = configure(api => {
   return async function findPeer (peerId, options = {}) {
@@ -18,11 +19,7 @@ module.exports = configure(api => {
     })
 
     for await (const data of res.ndjson()) {
-      if (data.Type === 3) {
-        throw new Error(data.Extra)
-      }
-
-      if (data.Type === 2 && data.Responses) {
+      if (data.Type === FinalPeer && data.Responses) {
         const { ID, Addrs } = data.Responses[0]
         return {
           id: ID,

packages/ipfs-http-client/src/dht/find-provs.js

@@ -4,6 +4,7 @@ const CID = require('cids')
 const multiaddr = require('multiaddr')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
+const { Provider } = require('./response-types')
 
 module.exports = configure(api => {
   return async function * findProvs (cid, options = {}) {
@@ -17,16 +18,7 @@ module.exports = configure(api => {
     })
 
     for await (const message of res.ndjson()) {
-      // 3 = QueryError
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
-      // https://github.com/libp2p/go-libp2p-kad-dht/blob/master/routing.go#L525-L526
-      if (message.Type === 3) {
-        throw new Error(message.Extra)
-      }
-
-      // 4 = Provider
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L20
-      if (message.Type === 4 && message.Responses) {
+      if (message.Type === Provider && message.Responses) {
         for (const { ID, Addrs } of message.Responses) {
           yield {
             id: ID,

packages/ipfs-http-client/src/dht/get.js

@@ -4,6 +4,7 @@ const { Buffer } = require('buffer')
 const encodeBufferURIComponent = require('../lib/encode-buffer-uri-component')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
+const { Value } = require('./response-types')
 
 module.exports = configure(api => {
   return async function get (key, options = {}) {
@@ -21,16 +22,7 @@ module.exports = configure(api => {
     })
 
     for await (const message of res.ndjson()) {
-      // 3 = QueryError
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
-      // https://github.com/ipfs/go-ipfs/blob/eb11f569b064b960d1aba4b5b8ca155a3bd2cb21/core/commands/dht.go#L472-L473
-      if (message.Type === 3) {
-        throw new Error(message.Extra)
-      }
-
-      // 5 = Value
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L21
-      if (message.Type === 5) {
+      if (message.Type === Value) {
         return message.Extra
       }
     }

packages/ipfs-http-client/src/dht/provide.js

@@ -20,13 +20,6 @@ module.exports = configure(api => {
     })
 
     for await (let message of res.ndjson()) {
-      // 3 = QueryError
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
-      // https://github.com/ipfs/go-ipfs/blob/eb11f569b064b960d1aba4b5b8ca155a3bd2cb21/core/commands/dht.go#L283-L284
-      if (message.Type === 3) {
-        throw new Error(message.Extra)
-      }
-
       message = toCamel(message)
       message.id = new CID(message.id)
       if (message.responses) {

packages/ipfs-http-client/src/dht/put.js

@@ -21,13 +21,6 @@ module.exports = configure(api => {
     })
 
     for await (let message of res.ndjson()) {
-      // 3 = QueryError
-      // https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L18
-      // https://github.com/ipfs/go-ipfs/blob/eb11f569b064b960d1aba4b5b8ca155a3bd2cb21/core/commands/dht.go#L472-L473
-      if (message.Type === 3) {
-        throw new Error(message.Extra)
-      }
-
       message = toCamel(message)
       message.id = new CID(message.id)
       if (message.responses) {

packages/ipfs-http-client/src/dht/response-types.js

@@ -0,0 +1,14 @@
+'use strict'
+
+// Response types are defined here:
+// https://github.com/libp2p/go-libp2p-core/blob/6e566d10f4a5447317a66d64c7459954b969bdab/routing/query.go#L15-L24
+module.exports = {
+  SendingQuery: 0,
+  PeerResponse: 1,
+  FinalPeer: 2,
+  QueryError: 3,
+  Provider: 4,
+  Value: 5,
+  AddingPeer: 6,
+  DialingPeer: 7
+}