feat: implement circuit v2

src/circuit/client.ts

@@ -359,7 +359,11 @@ export class RelayReservationManager extends EventEmitter<RelayReservationManage
         }
 
         const timeout = setTimeout(
-          (peerId: PeerId) => { void refreshReservation(peerId) },
+          (peerId: PeerId) => {
+            void refreshReservation(peerId).catch(err => {
+              log.error('error refreshing reservation for %p', peerId, err)
+            })
+          },
           Math.max(getExpiration(reservation.expire) - 100, 0),
           peerId
         )

src/circuit/transport.ts

@@ -126,7 +126,7 @@ export class Circuit implements Transport, Startable {
       hopTimeoutPromise.reject('timed out')
     }, this._init.hop.timeout)
     const pbstr = pbStream(stream)
-    const abortable = { value: pbstr, abort: (err: Error) => stream.abort(err) }
+
     try {
       const request: CircuitV2.HopMessage = await Promise.race([
         pbstr.pb(CircuitV2.HopMessage).read(),
@@ -140,7 +140,7 @@ export class Circuit implements Transport, Startable {
       await Promise.race([
         CircuitV2Handler.handleHopProtocol({
           connection,
-          stream: abortable,
+          stream: pbstr,
           connectionManager: this.components.connectionManager,
           relayPeer: this.components.peerId,
           relayAddrs: this.components.addressManager.getListenAddrs(),

src/circuit/v2/hop.ts

@@ -1,11 +1,11 @@
 import type { PeerId } from '@libp2p/interface-peer-id'
 import { RecordEnvelope } from '@libp2p/peer-record'
 import { logger } from '@libp2p/logger'
-import type { Connection } from '@libp2p/interface-connection'
+import type { Connection, Stream } from '@libp2p/interface-connection'
 import { HopMessage, Limit, Reservation, Status, StopMessage } from './pb/index.js'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import { multiaddr } from '@multiformats/multiaddr'
-import type { Acl, ReservationStore, Abortable } from './interfaces.js'
+import type { Acl, ReservationStore } from './interfaces.js'
 import { RELAY_V2_HOP_CODEC } from '../multicodec.js'
 import { stop } from './stop.js'
 import { ReservationVoucherRecord } from './reservation-voucher.js'
@@ -14,7 +14,6 @@ import type { ConnectionManager } from '@libp2p/interface-connection-manager'
 import type { ProtobufStream } from 'it-pb-stream'
 import { pbStream } from 'it-pb-stream'
 import { CIRCUIT_PROTO_CODE } from '../constants.js'
-import type { Uint8ArrayList } from 'uint8arraylist'
 import type { PeerStore } from '@libp2p/interface-peer-store'
 import { createLimitedRelay } from './util.js'
 import type { CodeError } from '@libp2p/interfaces/errors'
@@ -26,7 +25,7 @@ const RELAYED = 'relayed'
 export interface HopProtocolOptions {
   connection: Connection
   request: HopMessage
-  stream: Abortable<ProtobufStream<Uint8ArrayList | Uint8Array>>
+  stream: ProtobufStream<Stream>
   relayPeer: PeerId
   relayAddrs: Multiaddr[]
   limit?: Limit
@@ -44,7 +43,7 @@ export async function handleHopProtocol (options: HopProtocolOptions): Promise<v
     case HopMessage.Type.CONNECT: await handleConnect(options); break
     default: {
       log.error('invalid hop request type %s via peer %s', options.request.type, options.connection.remotePeer)
-      stream.value.pb(HopMessage).write({ type: HopMessage.Type.STATUS, status: Status.UNEXPECTED_MESSAGE })
+      stream.pb(HopMessage).write({ type: HopMessage.Type.STATUS, status: Status.UNEXPECTED_MESSAGE })
     }
   }
 }
@@ -75,7 +74,7 @@ export async function reserve (connection: Connection): Promise<Reservation> {
 
 const isRelayAddr = (ma: Multiaddr): boolean => ma.protoCodes().includes(CIRCUIT_PROTO_CODE)
 
-async function handleReserve ({ connection, stream: { value: pbstr }, relayPeer, relayAddrs, limit, acl, reservationStore, peerStore }: HopProtocolOptions): Promise<void> {
+async function handleReserve ({ connection, stream: pbstr, relayPeer, relayAddrs, limit, acl, reservationStore, peerStore }: HopProtocolOptions): Promise<void> {
   const hopstr = pbstr.pb(HopMessage)
   log('hop reserve request from %s', connection.remotePeer)
 
@@ -126,7 +125,7 @@ async function handleReserve ({ connection, stream: { value: pbstr }, relayPeer,
 
 async function handleConnect (options: HopProtocolOptions): Promise<void> {
   const { connection, stream, request, reservationStore, connectionManager, acl } = options
-  const hopstr = stream.value.pb(HopMessage)
+  const hopstr = stream.pb(HopMessage)
 
   log('hop connect request from %s', connection.remotePeer)
 
@@ -186,7 +185,7 @@ async function handleConnect (options: HopProtocolOptions): Promise<void> {
   }
 
   hopstr.write({ type: HopMessage.Type.STATUS, status: Status.OK })
-  const sourceStream = { ...stream, value: stream.value.unwrap() }
+  const sourceStream = stream.unwrap()
 
   log('connection to destination established, short circuiting streams...')
   const limit = (await reservationStore.get(dstPeer))?.limit

src/circuit/v2/interfaces.ts

@@ -1,7 +1,5 @@
 import type { PeerId } from '@libp2p/interface-peer-id'
 import type { Multiaddr } from '@multiformats/multiaddr'
-import type { Duplex } from 'it-stream-types'
-import type { Uint8ArrayList } from 'uint8arraylist'
 import type { Limit, Status } from './pb/index.js'
 
 export type ReservationStatus = Status.OK | Status.PERMISSION_DENIED | Status.RESERVATION_REFUSED
@@ -28,9 +26,3 @@ export interface Acl {
    */
   allowConnect: (src: PeerId, addr: Multiaddr, dst: PeerId) => Promise<AclStatus>
 }
-
-export type DuplexStream = Duplex<Uint8ArrayList, Uint8ArrayList | Uint8Array>
-export interface Abortable<T> {
-  value: T
-  abort: (err: Error) => void
-}

src/circuit/v2/stop.ts

@@ -1,20 +1,18 @@
 
 import { Status, StopMessage } from './pb/index.js'
-import type { Connection } from '@libp2p/interface-connection'
+import type { Connection, Stream } from '@libp2p/interface-connection'
 
 import { logger } from '@libp2p/logger'
 import { RELAY_V2_STOP_CODEC } from '../multicodec.js'
 import { multiaddr } from '@multiformats/multiaddr'
 import { pbStream, ProtobufStream } from 'it-pb-stream'
-import type { Uint8ArrayList } from 'uint8arraylist'
-import type { DuplexStream, Abortable } from './interfaces.js'
 
 const log = logger('libp2p:circuit:v2:stop')
 
 export interface HandleStopOptions {
   connection: Connection
   request: StopMessage
-  pbstr: ProtobufStream<Uint8ArrayList | Uint8Array>
+  pbstr: ProtobufStream<Stream>
 }
 
 const isValidStop = (request: StopMessage): boolean => {
@@ -66,7 +64,7 @@ export interface StopOptions {
 export async function stop ({
   connection,
   request
-}: StopOptions): Promise<Abortable<DuplexStream> | undefined> {
+}: StopOptions): Promise<Stream | undefined> {
   const stream = await connection.newStream([RELAY_V2_STOP_CODEC])
   log('starting circuit relay v2 stop request to %s', connection.remotePeer)
   const pbstr = pbStream(stream)
@@ -86,10 +84,7 @@ export async function stop ({
   }
   if (response.status === Status.OK) {
     log('stop request to %s was successful', connection.remotePeer)
-    return {
-      value: pbstr.unwrap(),
-      abort: (err: Error) => stream.abort(err)
-    }
+    return pbstr.unwrap()
   }
 
   log('stop request failed with code %d', response.status)

src/circuit/v2/util.ts

@@ -1,12 +1,12 @@
-import type { Sink, Source } from 'it-stream-types'
-import { Uint8ArrayList } from 'uint8arraylist'
+import type { Source } from 'it-stream-types'
+import type { Uint8ArrayList } from 'uint8arraylist'
 import type { Limit } from './pb/index.js'
 import { logger } from '@libp2p/logger'
-import type { DuplexStream, Abortable } from './interfaces.js'
+import type { Stream } from '@libp2p/interface-connection'
 
 const log = logger('libp2p:circuit:v2:util')
 
-const doRelay = (src: DuplexStream, dst: DuplexStream) => {
+const doRelay = (src: Stream, dst: Stream) => {
   queueMicrotask(() => {
     void dst.sink(src.source).catch(err => log.error('error while relating streams:', err))
   })
@@ -16,85 +16,102 @@ const doRelay = (src: DuplexStream, dst: DuplexStream) => {
   })
 }
 
-export function createLimitedRelay (source: Abortable<DuplexStream>, destination: Abortable<DuplexStream>, limit?: Limit) {
+export function createLimitedRelay (source: Stream, destination: Stream, limit?: Limit) {
   // trivial case
   if (limit == null) {
-    doRelay(source.value, destination.value)
+    doRelay(source, destination)
     return
   }
-  const dataLimit = limit.data ?? BigInt(0)
+
+  const dataLimit = limit.data ?? 0n
   const durationLimit = limit.duration ?? 0
   const src = durationLimitDuplex(dataLimitDuplex(source, dataLimit), durationLimit)
   const dst = durationLimitDuplex(dataLimitDuplex(destination, dataLimit), durationLimit)
-  doRelay(src.value, dst.value)
+
+  doRelay(src, dst)
 }
 
-const dataLimitSource = (source: Source<Uint8ArrayList>, abort: (err: Error) => void, limit: bigint): Source<Uint8ArrayList> => {
-  if (limit === BigInt(0)) {
-    return source
+const dataLimitSource = (stream: Stream, limit: bigint): Stream => {
+  if (limit === 0n) {
+    return stream
   }
 
-  return (async function * (): Source<Uint8ArrayList> {
-    let total = BigInt(0)
+  const source = stream.source
+
+  stream.source = (async function * (): Source<Uint8ArrayList> {
+    let total = 0n
+
     for await (const buf of source) {
-      const len = BigInt(buf.length)
+      const len = BigInt(buf.byteLength)
       if (total + len > limit) {
         log.error('attempted to send more data than limit: %s, resetting stream', limit.toString())
-        abort(new Error('exceeded connection data limit'))
+        stream.abort(new Error('exceeded connection data limit'))
         return
       }
-      total += len
+
       yield buf
+
+      total += len
     }
   })()
+
+  return stream
 }
 
-const adaptSource = (source: Source<Uint8ArrayList | Uint8Array>): Source<Uint8ArrayList> => (async function * () {
-  for await (const buf of source) {
-    if (buf instanceof Uint8Array) {
-      yield Uint8ArrayList.fromUint8Arrays([buf])
-    } else {
-      yield buf
-    }
+const dataLimitSink = (stream: Stream, limit: bigint): Stream => {
+  if (limit === 0n) {
+    return stream
   }
-})()
-
-const dataLimitSink = (sink: Sink<Uint8ArrayList | Uint8Array>, abort: (err: Error) => void, limit: bigint): Sink<Uint8ArrayList | Uint8Array> => {
-  return async (source: Source<Uint8ArrayList | Uint8Array>) => await sink(
-    dataLimitSource(
-      adaptSource(source),
-      abort,
-      limit
-    )
-  )
-}
 
-const dataLimitDuplex = (duplex: Abortable<DuplexStream>, limit: bigint): Abortable<DuplexStream> => {
-  return {
-    ...duplex,
-    value: {
-      ...duplex.value,
-      source: dataLimitSource(duplex.value.source, duplex.abort, limit),
-      sink: dataLimitSink(duplex.value.sink, duplex.abort, limit)
-    }
+  const sink = stream.sink
+
+  stream.sink = async (source: Source<Uint8ArrayList | Uint8Array>) => {
+    await sink((async function * (): Source<Uint8ArrayList | Uint8Array> {
+      let total = 0n
+
+      for await (const buf of source) {
+        const len = BigInt(buf.byteLength)
+        if (total + len > limit) {
+          log.error('attempted to send more data than limit: %s, resetting stream', limit.toString())
+          stream.abort(new Error('exceeded connection data limit'))
+          return
+        }
+
+        total += len
+        yield buf
+      }
+    })())
   }
+
+  return stream
+}
+
+const dataLimitDuplex = (stream: Stream, limit: bigint): Stream => {
+  dataLimitSource(stream, limit)
+  dataLimitSink(stream, limit)
+
+  return stream
 }
 
-const durationLimitDuplex = (duplex: Abortable<DuplexStream>, limit: number): Abortable<DuplexStream> => {
+const durationLimitDuplex = (stream: Stream, limit: number): Stream => {
   if (limit === 0) {
-    return duplex
+    return stream
   }
+
   let timedOut = false
   const timeout = setTimeout(
     () => {
       timedOut = true
-      duplex.abort(new Error('exceeded connection duration limit'))
+      stream.abort(new Error('exceeded connection duration limit'))
     },
     limit
   )
-  const source = (async function * (): Source<Uint8ArrayList> {
+
+  const source = stream.source
+
+  stream.source = (async function * (): Source<Uint8ArrayList> {
     try {
-      for await (const buf of duplex.value.source) {
+      for await (const buf of source) {
         if (timedOut) {
           return
         }
@@ -105,5 +122,5 @@ const durationLimitDuplex = (duplex: Abortable<DuplexStream>, limit: number): Ab
     }
   })()
 
-  return { ...duplex, value: { ...duplex.value, source } }
+  return stream
 }