fix(transport-webrtc): fix EOF race in readCandidatesUntilConnected (#3460)

* fix(transport-webrtc): fix UnexpectedEOFError race in readCandidatesUntilConnected

A race condition caused intermittent failures in the WebRTC 'simple'
transport compliance test: the listener closes its signaling stream just
before the initiator's connectionstatechange fires for the 'connected'
state. The initiator then receives EOF on the signaling stream and threw
an UnexpectedEOFError, discarding a perfectly good peer connection.

Root cause: connectedPromise was declared inside the try block, making it
inaccessible in the catch handler for stream read errors.

Fix: move connectedPromise outside the try block. In the catch handler,
if the PC is not yet 'connected', await connectedPromise — if it resolves
we ignore the stream error (the connection succeeded); if it rejects we
re-throw the original stream error.

* fix(transport-webrtc): bound signaling recovery by the dial signal

Race the wait for the connection state with the dial signal and surface the abort reason, reject early if the peer connection has already failed or closed, and move the tests to their own file.

Co-authored-by: Peerbit Bot <273107789+peerbit-org@users.noreply.github.com>

---------

Co-authored-by: tabcat <tabcat00@proton.me>
Co-authored-by: Peerbit Bot <273107789+peerbit-org@users.noreply.github.com>

Author: 3 people

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

@@ -1,6 +1,7 @@
 import { ConnectionFailedError, InvalidMessageError, InvalidMultiaddrError } from '@libp2p/interface'
 import { peerIdFromString } from '@libp2p/peer-id'
 import { CustomProgressEvent } from 'progress-events'
+import { raceSignal } from 'race-signal'
 import { RTCIceCandidate } from '../webrtc/index.ts'
 import { Message } from './pb/message.ts'
 import type { WebRTCDialEvents } from './transport.ts'
@@ -16,10 +17,10 @@ export interface ReadCandidatesOptions extends AbortOptions, LoggerOptions, Prog
 }
 
 export const readCandidatesUntilConnected = async (pc: RTCPeerConnection, stream: MessageStream<Message, Stream>, options: ReadCandidatesOptions): Promise<void> => {
-  try {
-    const connectedPromise = Promise.withResolvers<void>()
-    resolveOnConnected(pc, connectedPromise)
+  const connectedPromise = Promise.withResolvers<void>()
+  resolveOnConnected(pc, connectedPromise)
 
+  try {
     // read candidates until we are connected or we reach the end of the stream
     while (true) {
       // if we connect, stop trying to read from the stream
@@ -66,10 +67,16 @@ export const readCandidatesUntilConnected = async (pc: RTCPeerConnection, stream
       }
     }
   } catch (err) {
-    options.log.error('%s error parsing ICE candidate - %e', options.direction, err)
+    options.log.error('%s error reading ICE candidates - %e', options.direction, err)
 
-    if (options.signal?.aborted === true && pc.connectionState !== 'connected') {
-      throw err
+    // the signaling stream can end just before the peer connection connects
+    if (pc.connectionState !== 'connected') {
+      try {
+        await raceSignal(connectedPromise.promise, options.signal)
+      } catch {
+        options.signal?.throwIfAborted()
+        throw err
+      }
     }
   }
 }
@@ -80,6 +87,11 @@ function resolveOnConnected (pc: RTCPeerConnection, promise: DeferredPromise<voi
     return
   }
 
+  if (pc.connectionState === 'failed' || pc.connectionState === 'closed') {
+    promise.reject(new ConnectionFailedError(`RTCPeerConnection connection state was "${pc.connectionState}"`))
+    return
+  }
+
   pc.onconnectionstatechange = (_) => {
     switch (pc.connectionState) {
       case 'connected':

packages/transport-webrtc/test/read-candidates.spec.ts

@@ -0,0 +1,119 @@
+import { ConnectionFailedError } from '@libp2p/interface'
+import { defaultLogger } from '@libp2p/logger'
+import { pbStream, streamPair } from '@libp2p/utils'
+import { expect } from 'aegir/chai'
+import { Message } from '../src/private-to-private/pb/message.ts'
+import { readCandidatesUntilConnected } from '../src/private-to-private/util.ts'
+import type { Stream } from '@libp2p/interface'
+
+interface StubPeerConnection {
+  connectionState: RTCPeerConnectionState
+  onconnectionstatechange: ((evt: Event) => void) | null
+}
+
+function createPeerConnection (connectionState: RTCPeerConnectionState): StubPeerConnection {
+  return {
+    connectionState,
+    onconnectionstatechange: null
+  }
+}
+
+function setState (pc: StubPeerConnection, connectionState: RTCPeerConnectionState, ms: number): void {
+  setTimeout(() => {
+    pc.connectionState = connectionState
+    pc.onconnectionstatechange?.(new Event('connectionstatechange'))
+  }, ms)
+}
+
+async function readCandidates (pc: StubPeerConnection, stream: Stream, signal: AbortSignal = AbortSignal.timeout(1_000)): Promise<void> {
+  await readCandidatesUntilConnected(pc as any, pbStream(stream).pb(Message), {
+    direction: 'initiator',
+    signal,
+    log: defaultLogger().forComponent('test:webrtc')
+  })
+}
+
+describe('readCandidatesUntilConnected', () => {
+  let localStream: Stream
+  let remoteStream: Stream
+
+  beforeEach(async () => {
+    [localStream, remoteStream] = await streamPair()
+  })
+
+  afterEach(() => {
+    localStream.abort(new Error('test finished'))
+    remoteStream.abort(new Error('test finished'))
+  })
+
+  it('should resolve when the peer connection connects', async () => {
+    const pc = createPeerConnection('connecting')
+    setState(pc, 'connected', 50)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.undefined()
+  })
+
+  it('should resolve when the signaling stream ends before the peer connection connects', async () => {
+    const pc = createPeerConnection('connecting')
+    await remoteStream.close()
+    setState(pc, 'connected', 50)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.undefined()
+    expect(pc.connectionState).to.equal('connected')
+  })
+
+  it('should ignore a transient disconnected state', async () => {
+    const pc = createPeerConnection('connecting')
+    setState(pc, 'disconnected', 30)
+    setState(pc, 'connected', 60)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.undefined()
+  })
+
+  it('should reject when the peer connection fails', async () => {
+    const pc = createPeerConnection('connecting')
+    setState(pc, 'failed', 50)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.rejectedWith(ConnectionFailedError)
+  })
+
+  it('should reject when the peer connection closes', async () => {
+    const pc = createPeerConnection('connecting')
+    setState(pc, 'closed', 50)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.rejectedWith(ConnectionFailedError)
+  })
+
+  it('should reject when the peer connection has already failed', async () => {
+    const pc = createPeerConnection('failed')
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.rejectedWith(ConnectionFailedError)
+  })
+
+  it('should reject when the signaling stream ends and the peer connection fails', async () => {
+    const pc = createPeerConnection('connecting')
+    await remoteStream.close()
+    setState(pc, 'failed', 50)
+
+    await expect(readCandidates(pc, localStream)).to.eventually.be.rejected()
+  })
+
+  it('should reject with the abort reason when aborted while the peer connection is connecting', async () => {
+    const pc = createPeerConnection('connecting')
+    const controller = new AbortController()
+    const reason = new Error('dial cancelled')
+    setTimeout(() => { controller.abort(reason) }, 50)
+
+    await expect(readCandidates(pc, localStream, controller.signal)).to.eventually.be.rejectedWith(reason)
+  })
+
+  it('should reject with the abort reason when aborted after the signaling stream ends', async () => {
+    const pc = createPeerConnection('connecting')
+    await remoteStream.close()
+    const controller = new AbortController()
+    const reason = new Error('dial cancelled')
+    setTimeout(() => { controller.abort(reason) }, 50)
+
+    await expect(readCandidates(pc, localStream, controller.signal)).to.eventually.be.rejectedWith(reason)
+  })
+})