fix: upgrader should abort stream on close based on inbound limits (#1578)

Author: maschad

src/upgrader.ts

@@ -73,7 +73,6 @@ function findIncomingStreamLimit (protocol: string, registrar: Registrar): numbe
 function findOutgoingStreamLimit (protocol: string, registrar: Registrar): number | undefined {
   try {
     const { options } = registrar.getHandler(protocol)
-
     return options.maxOutboundStreams
   } catch (err: any) {
     if (err.code !== codes.ERR_NO_HANDLER_FOR_PROTOCOL) {
@@ -392,9 +391,10 @@ export class DefaultUpgrader implements Upgrader {
               const streamCount = countStreams(protocol, 'inbound', connection)
 
               if (streamCount === incomingLimit) {
-                muxedStream.abort(new CodeError(`Too many inbound protocol streams for protocol "${protocol}" - limit ${incomingLimit}`, codes.ERR_TOO_MANY_INBOUND_PROTOCOL_STREAMS))
+                const err = new CodeError(`Too many inbound protocol streams for protocol "${protocol}" - limit ${incomingLimit}`, codes.ERR_TOO_MANY_INBOUND_PROTOCOL_STREAMS)
+                muxedStream.abort(err)
 
-                return
+                throw err
               }
 
               // after the handshake the returned stream can have early data so override

test/upgrading/upgrader.spec.ts

@@ -34,6 +34,7 @@ import { DefaultComponents } from '../../src/components.js'
 import { StubbedInstance, stubInterface } from 'sinon-ts'
 import { yamux } from '@chainsafe/libp2p-yamux'
 import { EventEmitter } from '@libp2p/interfaces/events'
+import { logger } from '@libp2p/logger'
 
 const addrs = [
   multiaddr('/ip4/127.0.0.1/tcp/0'),
@@ -664,7 +665,10 @@ describe('libp2p.upgrader', () => {
         webSockets()
       ],
       streamMuxers: [
-        yamux()
+        yamux({
+          log: logger('yamux:local')
+        })
+        // mplex()
       ],
       connectionEncryption: [
         plaintext()
@@ -678,7 +682,10 @@ describe('libp2p.upgrader', () => {
         webSockets()
       ],
       streamMuxers: [
-        yamux()
+        yamux({
+          log: logger('yamux:remote')
+        })
+        // mplex()
       ],
       connectionEncryption: [
         plaintext()
@@ -726,8 +733,7 @@ describe('libp2p.upgrader', () => {
         webSockets()
       ],
       streamMuxers: [
-        yamux(),
-        mplex()
+        yamux()
       ],
       connectionEncryption: [
         plaintext()
@@ -741,8 +747,7 @@ describe('libp2p.upgrader', () => {
         webSockets()
       ],
       streamMuxers: [
-        yamux(),
-        mplex()
+        yamux()
       ],
       connectionEncryption: [
         plaintext()