fix: ensure streams are closed when protocol negotiation fails (#1236)

If an error is thrown during the initial stages of setting up a multiplexed stream, ensure we close the stream to free up any resources associated with it.

src/upgrader.ts

@@ -301,6 +301,10 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
             })
             .catch(err => {
               log.error(err)
+
+              if (muxedStream.timeline.close == null) {
+                muxedStream.close()
+              }
             })
         },
         // Run anytime a stream closes
@@ -330,6 +334,10 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
         } catch (err: any) {
           log.error('could not create new stream', err)
 
+          if (muxedStream.timeline.close == null) {
+            muxedStream.close()
+          }
+
           if (err.code != null) {
             throw err
           }

test/upgrading/upgrader.spec.ts

@@ -405,6 +405,29 @@ describe('Upgrader', () => {
     }))
       .to.eventually.be.rejected.with.property('code', 'ABORT_ERR')
   })
+
+  it('should close streams when protocol negotiation fails', async () => {
+    await remoteComponents.getRegistrar().unhandle('/echo/1.0.0')
+
+    const { inbound, outbound } = mockMultiaddrConnPair({ addrs, remotePeer })
+
+    const connections = await Promise.all([
+      localUpgrader.upgradeOutbound(outbound),
+      remoteUpgrader.upgradeInbound(inbound)
+    ])
+
+    expect(connections[0].streams).to.have.lengthOf(0)
+    expect(connections[1].streams).to.have.lengthOf(0)
+
+    await expect(connections[0].newStream('/echo/1.0.0'))
+      .to.eventually.be.rejected.with.property('code', 'ERR_UNSUPPORTED_PROTOCOL')
+
+    // wait for remote to close
+    await delay(100)
+
+    expect(connections[0].streams).to.have.lengthOf(0)
+    expect(connections[1].streams).to.have.lengthOf(0)
+  })
 })
 
 describe('libp2p.upgrader', () => {