fix: delay notification of early WebRTC stream creation (#2206)

The datachannel muxer is created during set up of the `Connection` object.  If we notify of early stream creation before the `Connection` object is properly configured, the early streams will be lost.

This can happen when the remote opens a data channel before the local node has finished setting up it's end of the connection.

The fix is to notify asynchronously which gives the upgrader enough time to finish setting up the `Connection`.

Author: achingbrain

packages/transport-webrtc/src/muxer.ts

@@ -137,14 +137,24 @@ export class DataChannelMuxer implements StreamMuxer {
       init?.onIncomingStream?.(stream)
     }
 
-    this.init.streams.forEach(bufferedStream => {
-      bufferedStream.onEnd = () => {
-        this.#onStreamEnd(bufferedStream.stream, bufferedStream.channel)
-      }
-
-      this.metrics?.increment({ incoming_stream: true })
-      this.init?.onIncomingStream?.(bufferedStream.stream)
-    })
+    // the DataChannelMuxer constructor is called during set up of the
+    // connection by the upgrader.
+    //
+    // If we invoke `init.onIncomingStream` immediately, the connection object
+    // will not be set up yet so add a tiny delay before letting the
+    // connection know about early streams
+    if (this.init.streams.length > 0) {
+      queueMicrotask(() => {
+        this.init.streams.forEach(bufferedStream => {
+          bufferedStream.onEnd = () => {
+            this.#onStreamEnd(bufferedStream.stream, bufferedStream.channel)
+          }
+
+          this.metrics?.increment({ incoming_stream: true })
+          this.init?.onIncomingStream?.(bufferedStream.stream)
+        })
+      })
+    }
   }
 
   #onStreamEnd (stream: Stream, channel: RTCDataChannel): void {

packages/transport-webrtc/src/private-to-private/transport.ts

@@ -121,6 +121,10 @@ export class WebRTCTransport implements Transport, Startable {
     log.trace('dialing address: %a', ma)
 
     const peerConnection = new RTCPeerConnection(this.init.rtcConfiguration)
+    const muxerFactory = new DataChannelMuxerFactory({
+      peerConnection,
+      dataChannelOptions: this.init.dataChannel
+    })
 
     const { remoteAddress } = await initiateConnection({
       peerConnection,
@@ -141,10 +145,7 @@ export class WebRTCTransport implements Transport, Startable {
     const connection = await options.upgrader.upgradeOutbound(webRTCConn, {
       skipProtection: true,
       skipEncryption: true,
-      muxerFactory: new DataChannelMuxerFactory({
-        peerConnection,
-        dataChannelOptions: this.init.dataChannel
-      })
+      muxerFactory
     })
 
     // close the connection on shut down
@@ -156,6 +157,10 @@ export class WebRTCTransport implements Transport, Startable {
   async _onProtocol ({ connection, stream }: IncomingStreamData): Promise<void> {
     const signal = AbortSignal.timeout(this.init.inboundConnectionTimeout ?? INBOUND_CONNECTION_TIMEOUT)
     const peerConnection = new RTCPeerConnection(this.init.rtcConfiguration)
+    const muxerFactory = new DataChannelMuxerFactory({
+      peerConnection,
+      dataChannelOptions: this.init.dataChannel
+    })
 
     try {
       const { remoteAddress } = await handleIncomingStream({
@@ -178,7 +183,7 @@ export class WebRTCTransport implements Transport, Startable {
       await this.components.upgrader.upgradeInbound(webRTCConn, {
         skipEncryption: true,
         skipProtection: true,
-        muxerFactory: new DataChannelMuxerFactory({ peerConnection, dataChannelOptions: this.init.dataChannel })
+        muxerFactory
       })
 
       // close the stream if SDP messages have been exchanged successfully

packages/transport-webrtc/test/muxer.spec.ts

@@ -0,0 +1,44 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions */
+
+import { expect } from 'aegir/chai'
+import pRetry from 'p-retry'
+import { stubInterface } from 'sinon-ts'
+import { DataChannelMuxerFactory } from '../src/muxer.js'
+
+describe('muxer', () => {
+  it('should delay notification of early streams', async () => {
+    let onIncomingStreamInvoked = false
+
+    // @ts-expect-error incomplete implementation
+    const peerConnection: RTCPeerConnection = {}
+
+    const muxerFactory = new DataChannelMuxerFactory({
+      peerConnection
+    })
+
+    // simulate early connection
+    // @ts-expect-error incomplete implementation
+    const event: RTCDataChannelEvent = {
+      channel: stubInterface<RTCDataChannel>({
+        readyState: 'connecting'
+      })
+    }
+    peerConnection.ondatachannel?.(event)
+
+    muxerFactory.createStreamMuxer({
+      onIncomingStream: () => {
+        onIncomingStreamInvoked = true
+      }
+    })
+
+    expect(onIncomingStreamInvoked).to.be.false()
+
+    await pRetry(() => {
+      if (!onIncomingStreamInvoked) {
+        throw new Error('onIncomingStreamInvoked was still false')
+      }
+    })
+
+    expect(onIncomingStreamInvoked).to.be.true()
+  })
+})