Merge branch 'main' into provider-validty

packages/integration-tests/test/circuit-relay.node.ts

@@ -519,8 +519,8 @@ describe('circuit-relay', () => {
       await deferred.promise
 
       // should have closed connections to remote and to relay
-      expect(events[0].detail.remotePeer.toString()).to.equal(relay1.peerId.toString())
-      expect(events[1].detail.remotePeer.toString()).to.equal(remote.peerId.toString())
+      expect(events[0].detail.remotePeer.toString()).to.equal(remote.peerId.toString())
+      expect(events[1].detail.remotePeer.toString()).to.equal(relay1.peerId.toString())
     })
 
     it('should mark an outgoing relayed connection as limited', async () => {

packages/protocol-ping/src/ping.ts

@@ -1,5 +1,5 @@
 import { randomBytes } from '@libp2p/crypto'
-import { ProtocolError, TimeoutError } from '@libp2p/interface'
+import { ProtocolError, TimeoutError, setMaxListeners } from '@libp2p/interface'
 import { byteStream } from 'it-byte-stream'
 import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
 import { PROTOCOL_PREFIX, PROTOCOL_NAME, PING_LENGTH, PROTOCOL_VERSION, TIMEOUT, MAX_INBOUND_STREAMS, MAX_OUTBOUND_STREAMS } from './constants.js'
@@ -60,10 +60,12 @@ export class PingService implements Startable, PingServiceInterface {
     const { stream } = data
     const start = Date.now()
     const bytes = byteStream(stream)
+    let pinged = false
 
     Promise.resolve().then(async () => {
       while (true) {
         const signal = AbortSignal.timeout(this.timeout)
+        setMaxListeners(Infinity, signal)
         signal.addEventListener('abort', () => {
           stream?.abort(new TimeoutError('ping timeout'))
         })
@@ -74,15 +76,34 @@ export class PingService implements Startable, PingServiceInterface {
         await bytes.write(buf, {
           signal
         })
+
+        pinged = true
       }
     })
       .catch(err => {
-        this.log.error('incoming ping from %p failed with error', data.connection.remotePeer, err)
+        // ignore the error if we've processed at least one ping, the remote
+        // closed the stream and we handled or are handling the close cleanly
+        if (pinged && err.name === 'UnexpectedEOFError' && stream.readStatus !== 'ready') {
+          return
+        }
+
+        this.log.error('incoming ping from %p failed with error - %e', data.connection.remotePeer, err)
         stream?.abort(err)
       })
       .finally(() => {
         const ms = Date.now() - start
         this.log('incoming ping from %p complete in %dms', data.connection.remotePeer, ms)
+
+        const signal = AbortSignal.timeout(this.timeout)
+        setMaxListeners(Infinity, signal)
+
+        stream.close({
+          signal
+        })
+          .catch(err => {
+            this.log.error('error closing ping stream from %p - %e', data.connection.remotePeer, err)
+            stream?.abort(err)
+          })
       })
   }
 

packages/upnp-nat/src/upnp-nat.ts

@@ -1,6 +1,6 @@
 import { upnpNat } from '@achingbrain/nat-port-mapper'
 import { isIPv4, isIPv6 } from '@chainsafe/is-ip'
-import { InvalidParametersError, serviceCapabilities, start, stop } from '@libp2p/interface'
+import { InvalidParametersError, serviceCapabilities, serviceDependencies, start, stop } from '@libp2p/interface'
 import { debounce } from '@libp2p/utils/debounce'
 import { isLoopback } from '@libp2p/utils/multiaddr/is-loopback'
 import { isPrivate } from '@libp2p/utils/multiaddr/is-private'
@@ -87,6 +87,16 @@ export class UPnPNAT implements Startable, UPnPNATInterface {
     '@libp2p/nat-traversal'
   ]
 
+  get [serviceDependencies] (): string[] {
+    if (!this.autoConfirmAddress) {
+      return [
+        '@libp2p/autonat'
+      ]
+    }
+
+    return []
+  }
+
   isStarted (): boolean {
     return this.started
   }

packages/utils/src/abstract-stream.ts

@@ -284,6 +284,10 @@ export abstract class AbstractStream implements Stream {
 
   // Close for both Reading and Writing
   async close (options?: AbortOptions): Promise<void> {
+    if (this.status !== 'open') {
+      return
+    }
+
     this.log.trace('closing gracefully')
 
     this.status = 'closing'