feat!: specify stream limits (#348)

Adds two config params `maxInboundStreams` and `maxOutboundStreams` that allow us to configure how many protocol streams are allowed to be running in parallel.

Initial default values are completely arbitrary, will need to tune these for typical workloads.

BREAKING CHANGE: Updates to new registrar API

Author: achingbrain

package.json

@@ -133,22 +133,22 @@
     "release": "aegir release"
   },
   "dependencies": {
-    "@libp2p/components": "^1.0.0",
+    "@libp2p/components": "^2.0.0",
     "@libp2p/crypto": "^1.0.0",
     "@libp2p/interface-address-manager": "^1.0.1",
-    "@libp2p/interface-connection": "^1.0.1",
+    "@libp2p/interface-connection": "^2.0.0",
     "@libp2p/interface-connection-manager": "^1.0.0",
     "@libp2p/interface-dht": "^1.0.0",
     "@libp2p/interface-peer-discovery": "^1.0.0",
     "@libp2p/interface-peer-id": "^1.0.2",
     "@libp2p/interface-peer-info": "^1.0.1",
     "@libp2p/interface-peer-store": "^1.0.0",
-    "@libp2p/interface-registrar": "^1.0.0",
+    "@libp2p/interface-registrar": "^2.0.0",
     "@libp2p/interfaces": "^3.0.2",
-    "@libp2p/logger": "^1.1.6",
+    "@libp2p/logger": "^2.0.0",
     "@libp2p/peer-id": "^1.1.13",
     "@libp2p/record": "^2.0.0",
-    "@libp2p/topology": "^2.0.0",
+    "@libp2p/topology": "^3.0.0",
     "@multiformats/multiaddr": "^10.1.5",
     "abortable-iterator": "^4.0.2",
     "any-signal": "^3.0.0",
@@ -178,9 +178,9 @@
     "varint": "^6.0.0"
   },
   "devDependencies": {
-    "@libp2p/interface-mocks": "^1.0.1",
+    "@libp2p/interface-mocks": "^2.0.0",
     "@libp2p/peer-id-factory": "^1.0.9",
-    "@libp2p/peer-store": "^2.0.0",
+    "@libp2p/peer-store": "^3.0.0",
     "@types/lodash.random": "^3.2.6",
     "@types/lodash.range": "^3.2.6",
     "@types/varint": "^6.0.0",

src/index.ts

@@ -45,6 +45,16 @@ export interface KadDHTInit {
    * be evicted from the routing table or not (default 10)
    */
   pingConcurrency?: number
+
+  /**
+   * How many parallel incoming streams to allow on the DHT protocol per-connection
+   */
+  maxInboundStreams?: number
+
+  /**
+   * How many parallel outgoing streams to allow on the DHT protocol per-connection
+   */
+  maxOutboundStreams?: number
 }
 
 export class KadDHT extends DualKadDHT {

src/kad-dht.ts

@@ -27,6 +27,9 @@ import { selectors as recordSelectors } from '@libp2p/record/selectors'
 import { symbol } from '@libp2p/interface-peer-discovery'
 import { PROTOCOL_DHT, PROTOCOL_PREFIX, LAN_PREFIX } from './constants.js'
 
+export const DEFAULT_MAX_INBOUND_STREAMS = 32
+export const DEFAULT_MAX_OUTBOUND_STREAMS = 64
+
 export interface SingleKadDHTInit extends KadDHTInit {
   /**
    * Whether to start up in lan or wan mode
@@ -60,6 +63,8 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
   private readonly rpc: RPC
   private readonly topologyListener: TopologyListener
   private readonly querySelf: QuerySelf
+  private readonly maxInboundStreams: number
+  private readonly maxOutboundStreams: number
 
   /**
    * Create a new KadDHT
@@ -76,7 +81,9 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
       lan,
       protocolPrefix,
       pingTimeout,
-      pingConcurrency
+      pingConcurrency,
+      maxInboundStreams,
+      maxOutboundStreams
     } = init
 
     this.running = false
@@ -85,6 +92,8 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
     this.protocol = `${protocolPrefix ?? PROTOCOL_PREFIX}${lan === true ? LAN_PREFIX : ''}${PROTOCOL_DHT}`
     this.kBucketSize = kBucketSize ?? 20
     this.clientMode = clientMode ?? true
+    this.maxInboundStreams = maxInboundStreams ?? DEFAULT_MAX_INBOUND_STREAMS
+    this.maxOutboundStreams = maxOutboundStreams ?? DEFAULT_MAX_OUTBOUND_STREAMS
     this.routingTable = new RoutingTable({
       kBucketSize,
       lan: this.lan,
@@ -264,7 +273,10 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT, In
     } else {
       this.log('enabling server mode')
       this.clientMode = false
-      await this.components.getRegistrar().handle(this.protocol, this.rpc.onIncomingStream.bind(this.rpc))
+      await this.components.getRegistrar().handle(this.protocol, this.rpc.onIncomingStream.bind(this.rpc), {
+        maxInboundStreams: this.maxInboundStreams,
+        maxOutboundStreams: this.maxOutboundStreams
+      })
     }
   }
 

src/network.ts

@@ -97,8 +97,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
 
     try {
       const connection = await this.components.getConnectionManager().openConnection(to, options)
-      const streamData = await connection.newStream(this.protocol, options)
-      stream = streamData.stream
+      const stream = await connection.newStream(this.protocol, options)
 
       const response = await this._writeReadMessage(stream, msg.serialize(), options)
 
@@ -134,8 +133,7 @@ export class Network extends EventEmitter<NetworkEvents> implements Startable, I
 
     try {
       const connection = await this.components.getConnectionManager().openConnection(to, options)
-      const data = await connection.newStream(this.protocol, options)
-      stream = data.stream
+      const stream = await connection.newStream(this.protocol, options)
 
       await this._writeMessage(stream, msg.serialize(), options)
 

src/routing-table/index.ts

@@ -158,7 +158,7 @@ export class RoutingTable implements Startable, Initializable {
 
               this.log('pinging old contact %p', oldContact.peer)
               const connection = await this.components.getConnectionManager().openConnection(oldContact.peer, options)
-              const { stream } = await connection.newStream(this.protocol, options)
+              const stream = await connection.newStream(this.protocol, options)
               stream.close()
               responded++
             } catch (err: any) {

test/network.spec.ts

@@ -80,9 +80,14 @@ describe('Network', () => {
               finish()
             }
 
+            const stream = mockStream({ source, sink })
+
             return {
-              protocol,
-              stream: mockStream({ source, sink })
+              ...stream,
+              stat: {
+                ...stream.stat,
+                protocol
+              }
             }
           }
         }

test/routing-table.spec.ts

@@ -126,7 +126,7 @@ describe('Routing Table', () => {
     table.kb.add(oldPeer)
 
     // simulate connection succeeding
-    const newStreamStub = sinon.stub().withArgs(PROTOCOL_DHT).resolves({ stream: { close: sinon.stub() } })
+    const newStreamStub = sinon.stub().withArgs(PROTOCOL_DHT).resolves({ close: sinon.stub() })
     const openConnectionStub = sinon.stub().withArgs(oldPeer.peer).resolves({
       newStream: newStreamStub
     })

test/rpc/index.spec.ts

@@ -93,7 +93,6 @@ describe('rpc', () => {
     }
 
     await rpc.onIncomingStream({
-      protocol: 'protocol',
       stream: mockStream(duplexStream),
       connection: await mockConnection(mockMultiaddrConnection(duplexStream, otherPeerId))
     })