chore: fix up tests

libp2p · achingbrain · Feb 23, 2024 · Apr 3, 2024 · Apr 3, 2024 · Apr 3, 2024
commit 016da0afdfef8dc70b9c1b807f8c87a64d965454
@@ -59,6 +59,7 @@
     "@libp2p/peer-id": "^4.0.8",
     "@multiformats/multiaddr": "^12.2.1",
     "@multiformats/multiaddr-matcher": "^1.2.0",
+    "browser-readablestream-to-it": "^2.0.5",
     "it-stream-types": "^2.0.1",
     "multiformats": "^13.1.0",
     "race-signal": "^1.0.2",

@@ -32,6 +32,7 @@
 import { noise } from '@chainsafe/libp2p-noise'
 import { AbortError, transportSymbol } from '@libp2p/interface'
 import { WebTransport as WebTransportMatcher } from '@multiformats/multiaddr-matcher'
+import { raceSignal } from 'race-signal'
 import createListener from './listener.js'
 import { webtransportMuxer } from './muxer.js'
 import { inertDuplex } from './utils/inert-duplex.js'
@@ -104,8 +105,8 @@

  async dial (ma: Multiaddr, options: DialOptions): Promise<Connection> {
    if (options?.signal?.aborted === true) {
      throw new AbortError()
    }

    this.log('dialing %s', ma)
    const localPeer = this.components.peerId
@@ -193,13 +194,13 @@
           cleanUpWTSession('remote_close')
         })
 
-      if (!await this.authenticateWebTransport(wt, localPeer, remotePeer, certhashes)) {
+      if (!await raceSignal(this.authenticateWebTransport(wt, localPeer, remotePeer, certhashes), options.signal)) {
         throw new Error('Failed to authenticate webtransport')
       }
 
      if (options?.signal?.aborted === true) {
        throw new AbortError()
      }

      this.metrics?.dialerEvents.increment({ open: true })

@@ -300,10 +301,10 @@
  }

  createListener (options: CreateListenerOptions): Listener {
    return createListener(this.components, {
      ...options,
      certificates: this.config.certificates
    })
  }

  /**

diff --git a/packages/transport-webtransport/src/stream.ts b/packages/transport-webtransport/src/stream.ts
@@ -19,15 +19,19 @@ class WebTransportStream extends AbstractStream {
 
     Promise.resolve().then(async () => {
       while (true) {
+        init.log('readData waiting for reader to be ready')
         const result = await this.reader.read()
 
+        init.log('readData read', result)
+
         if (result.done) {
-          init.log('remote closed read')
-          this.remoteCloseRead()
+          init.log('remote closed write')
+          this.remoteCloseWrite()
           return
         }
 
         if (result.value != null) {
+          init.log('push valud', new Uint8ArrayList(result.value))
           this.sourcePush(new Uint8ArrayList(result.value))
         }
       }
@@ -45,7 +49,7 @@ class WebTransportStream extends AbstractStream {
         init.log('writer close promise rejected', err)
       })
       .finally(() => {
-        this.remoteCloseWrite()
+        this.remoteCloseRead()
       })
   }
 
@@ -55,7 +59,9 @@ class WebTransportStream extends AbstractStream {
 
   async sendData (buf: Uint8ArrayList, options?: AbortOptions): Promise<void> {
     for await (const chunk of buf) {
+      this.log('sendData waiting for writer to be ready')
       await raceSignal(this.writer.ready, options?.signal)
+      this.log('sendData sending data', chunk)
       await raceSignal(this.writer.write(chunk), options?.signal)
     }
   }

@@ -7,6 +7,7 @@ import { expect } from 'aegir/chai'
 import map from 'it-map'
 import toBuffer from 'it-to-buffer'
 import { createLibp2p, type Libp2p } from 'libp2p'
+import pWaitFor from 'p-wait-for'
 import { webTransport } from '../src/index.js'
 import { randomBytes } from './fixtures/random-bytes.js'
 
@@ -19,6 +20,9 @@ describe('libp2p-webtransport', () => {
       connectionEncryption: [noise()],
       connectionGater: {
         denyDialMultiaddr: async () => false
+      },
+      connectionManager: {
+        minConnections: 0
       }
     })
   })
@@ -111,7 +115,7 @@ describe('libp2p-webtransport', () => {
     expect(err.toString()).to.contain('aborted')
   })
 
-  it('connects to ipv6 addresses', async function () {
+  it.skip('connects to ipv6 addresses', async function () {
     if (process.env.disableIp6 === 'true') {
       return this.skip()
     }
@@ -153,6 +157,13 @@ describe('libp2p-webtransport', () => {
 
     await stream.sink(gen())
 
+    // closing takes a little bit of time
+    await pWaitFor(() => {
+      return stream.writeStatus === 'closed'
+    }, {
+      interval: 100
+    })
+
     expect(stream.writeStatus).to.equal('closed')
     expect(stream.timeline.closeWrite).to.be.greaterThan(0)