fix: time out slow senders (#455)

If no Bitswap message is received from a remote after 30s, close the stream.

The timeout value is configurable.

Author: achingbrain

package.json

@@ -168,6 +168,7 @@
     "@libp2p/tracked-map": "^2.0.0",
     "@multiformats/multiaddr": "^10.1.8",
     "@vascosantos/moving-average": "^1.1.0",
+    "abortable-iterator": "^4.0.2",
     "any-signal": "^3.0.0",
     "blockstore-core": "^1.0.2",
     "debug": "^4.2.0",
@@ -179,6 +180,7 @@
     "multiformats": "^9.0.4",
     "protobufjs": "^6.10.2",
     "readable-stream": "^4.0.0",
+    "timeout-abort-controller": "^3.0.0",
     "uint8arrays": "^3.0.0",
     "varint": "^6.0.0",
     "varint-decoder": "^1.0.0"

src/bitswap.js

@@ -51,6 +51,7 @@ export class Bitswap extends BaseBlockstore {
    * @param {number} [options.statsComputeThrottleMaxQueueSize=1000]
    * @param {number} [options.maxInboundStreams=32]
    * @param {number} [options.maxOutboundStreams=32]
+   * @param {number} [options.incomingStreamTimeout=30000]
    * @param {MultihashHasherLoader} [options.hashLoader]
    */
   constructor (libp2p, blockstore, options = {}) {
@@ -72,7 +73,8 @@ export class Bitswap extends BaseBlockstore {
     this.network = new Network(libp2p, this, this._stats, {
       hashLoader: options.hashLoader,
       maxInboundStreams: options.maxInboundStreams,
-      maxOutboundStreams: options.maxOutboundStreams
+      maxOutboundStreams: options.maxOutboundStreams,
+      incomingStreamTimeout: options.incomingStreamTimeout
     })
 
     // local database

src/index.js

@@ -19,6 +19,7 @@ import { Bitswap } from './bitswap.js'
  * @param {number} [options.statsComputeThrottleMaxQueueSize=1000]
  * @param {number} [options.maxInboundStreams=32]
  * @param {number} [options.maxOutboundStreams=128]
+ * @param {number} [options.incomingStreamTimeout=30000]
  * @param {MultihashHasherLoader} [options.hashLoader]
  * @returns {IPFSBitswap}
  */

src/network.js

@@ -4,6 +4,8 @@ import { createTopology } from '@libp2p/topology'
 import { BitswapMessage as Message } from './message/index.js'
 import * as CONSTANTS from './constants.js'
 import { logger } from './utils/index.js'
+import { TimeoutController } from 'timeout-abort-controller'
+import { abortableSource } from 'abortable-iterator'
 
 /**
  * @typedef {import('@libp2p/interface-peer-id').PeerId} PeerId
@@ -26,6 +28,7 @@ const BITSWAP120 = '/ipfs/bitswap/1.2.0'
 
 const DEFAULT_MAX_INBOUND_STREAMS = 32
 const DEFAULT_MAX_OUTBOUND_STREAMS = 128
+const DEFAULT_INCOMING_STREAM_TIMEOUT = 30000
 
 export class Network {
   /**
@@ -37,6 +40,7 @@ export class Network {
    * @param {MultihashHasherLoader} [options.hashLoader]
    * @param {number} [options.maxInboundStreams=32]
    * @param {number} [options.maxOutboundStreams=32]
+   * @param {number} [options.incomingStreamTimeout=30000]
    */
   constructor (libp2p, bitswap, stats, options = {}) {
     this._log = logger(libp2p.peerId, 'network')
@@ -60,6 +64,7 @@ export class Network {
     this._hashLoader = options.hashLoader
     this._maxInboundStreams = options.maxInboundStreams ?? DEFAULT_MAX_INBOUND_STREAMS
     this._maxOutboundStreams = options.maxOutboundStreams ?? DEFAULT_MAX_OUTBOUND_STREAMS
+    this._incomingStreamTimeout = options.incomingStreamTimeout ?? DEFAULT_INCOMING_STREAM_TIMEOUT
   }
 
   async start () {
@@ -117,11 +122,13 @@ export class Network {
       return
     }
 
+    const controller = new TimeoutController(this._incomingStreamTimeout)
+
     Promise.resolve().then(async () => {
       this._log('incoming new bitswap %s connection from %p', stream.stat.protocol, connection.remotePeer)
 
       await pipe(
-        stream,
+        abortableSource(stream.source, controller.signal),
         lp.decode(),
         async (source) => {
           for await (const data of source) {
@@ -138,6 +145,10 @@ export class Network {
     })
       .catch(err => {
         this._log(err)
+        stream.abort(err)
+      })
+      .finally(() => {
+        controller.clear()
       })
   }
 

test/network/network.node.js

@@ -11,6 +11,7 @@ import { BitswapMessage as Message } from '../../src/message/index.js'
 import { Stats } from '../../src/stats/index.js'
 import sinon from 'sinon'
 import { CID } from 'multiformats/cid'
+import delay from 'delay'
 
 /**
  * @typedef {import('libp2p').Libp2p} Libp2p
@@ -338,4 +339,41 @@ describe('network', () => {
     expect(mockDial.calledWith(provider1.id)).to.be.true()
     expect(mockDial.calledWith(provider2.id)).to.be.true()
   })
+
+  it('times out slow senders', async () => {
+    const deferred = pDefer()
+
+    const libp2p = {
+      handle: sinon.stub(),
+      registrar: {
+        register: sinon.stub()
+      },
+      getConnections: () => []
+    }
+
+    // @ts-expect-error not a complete libp2p implementation
+    const network = new Network(libp2p, {}, {}, {
+      incomingStreamTimeout: 1
+    })
+    await network.start()
+
+    const stream = {
+      source: (async function * () {
+        await delay(100)
+        yield 'hello'
+      }()),
+      abort: (/** @type {Error} **/ err) => {
+        deferred.resolve(err)
+      },
+      stat: {
+        protocol: 'hello'
+      }
+    }
+
+    const handler = libp2p.handle.getCall(0).args[1]
+    handler({ stream, connection: {} })
+
+    const err = await deferred.promise
+    expect(err).to.have.property('code', 'ABORT_ERR')
+  })
 })