feat: add config validation to perf service

Author: maschad

packages/libp2p/src/connection-manager/utils.ts

@@ -3,7 +3,7 @@ import { type AbortOptions, multiaddr, type Multiaddr } from '@multiformats/mult
 import { type ClearableSignal, anySignal } from 'any-signal'
 import { type ObjectSchema, array, number, object, string } from 'yup'
 import { validateMultiaddr } from '../config/helpers.js'
-import { AUTO_DIAL_CONCURRENCY, AUTO_DIAL_INTERVAL, AUTO_DIAL_PRIORITY, DIAL_TIMEOUT, INBOUND_CONNECTION_THRESHOLD, INBOUND_UPGRADE_TIMEOUT, MAX_CONNECTIONS, MAX_INCOMING_PENDING_CONNECTIONS, MAX_PARALLEL_DIALS, MAX_PARALLEL_DIALS_PER_PEER, MAX_PEER_ADDRS_TO_DIAL, MIN_CONNECTIONS } from './constants.js'
+import { AUTO_DIAL_CONCURRENCY, AUTO_DIAL_INTERVAL, AUTO_DIAL_PRIORITY, DIAL_TIMEOUT, INBOUND_CONNECTION_THRESHOLD, INBOUND_UPGRADE_TIMEOUT, MAX_CONNECTIONS, MAX_INCOMING_PENDING_CONNECTIONS, MAX_PARALLEL_DIALS, MAX_PEER_ADDRS_TO_DIAL, MIN_CONNECTIONS } from './constants.js'
 import type { ConnectionManagerInit } from '.'
 import type { LoggerOptions } from '@libp2p/interface'
 

packages/protocol-fetch/src/fetch.ts

@@ -22,8 +22,8 @@ import type { IncomingStreamData } from '@libp2p/interface-internal/registrar'
 
 const configValidator = object({
   timeout: number().integer().default(TIMEOUT),
-  maxInboundStreams: number().integer().default(MAX_INBOUND_STREAMS),
-  maxOutboundStreams: number().integer().default(MAX_OUTBOUND_STREAMS)
+  maxInboundStreams: number().integer().min(0).default(MAX_INBOUND_STREAMS),
+  maxOutboundStreams: number().integer().min(0).default(MAX_OUTBOUND_STREAMS)
 })
 export class Fetch implements Startable, FetchInterface {
   public readonly protocol: string

packages/protocol-perf/package.json

@@ -51,7 +51,8 @@
     "@libp2p/interface": "^0.1.6",
     "@libp2p/interface-internal": "^0.1.9",
     "@multiformats/multiaddr": "^12.1.10",
-    "it-pushable": "^3.2.3"
+    "it-pushable": "^3.2.3",
+    "yup": "^1.3.2"
   },
   "devDependencies": {
     "@libp2p/interface-compliance-tests": "^4.1.5",

packages/protocol-perf/src/perf-service.ts

@@ -1,11 +1,20 @@
 import { pushable } from 'it-pushable'
+import { boolean, number, object, string } from 'yup'
 import { MAX_INBOUND_STREAMS, MAX_OUTBOUND_STREAMS, PROTOCOL_NAME, RUN_ON_TRANSIENT_CONNECTION, WRITE_BLOCK_SIZE } from './constants.js'
 import type { PerfOptions, PerfOutput, PerfComponents, PerfInit, Perf as PerfInterface } from './index.js'
 import type { Logger } from '@libp2p/interface'
 import type { Startable } from '@libp2p/interface/startable'
 import type { IncomingStreamData } from '@libp2p/interface-internal/registrar'
 import type { Multiaddr } from '@multiformats/multiaddr'
 
+const configValidator = object({
+  maxInboundStreams: number().integer().min(0).default(MAX_INBOUND_STREAMS),
+  maxOutboundStreams: number().integer().min(0).default(MAX_OUTBOUND_STREAMS),
+  protocolName: string().default(PROTOCOL_NAME),
+  writeBlockSize: number().integer().min(1).default(WRITE_BLOCK_SIZE),
+  runOnTransientConnection: boolean().default(RUN_ON_TRANSIENT_CONNECTION)
+})
+
 export class Perf implements Startable, PerfInterface {
   private readonly log: Logger
   public readonly protocol: string
@@ -21,12 +30,15 @@ export class Perf implements Startable, PerfInterface {
     this.components = components
     this.log = components.logger.forComponent('libp2p:perf')
     this.started = false
-    this.protocol = init.protocolName ?? PROTOCOL_NAME
-    this.writeBlockSize = init.writeBlockSize ?? WRITE_BLOCK_SIZE
+
+    const config = configValidator.validateSync(init)
+
+    this.protocol = config.protocolName
+    this.writeBlockSize = config.writeBlockSize
+    this.maxInboundStreams = config.maxInboundStreams
+    this.maxOutboundStreams = config.maxOutboundStreams
+    this.runOnTransientConnection = config.runOnTransientConnection
     this.databuf = new ArrayBuffer(this.writeBlockSize)
-    this.maxInboundStreams = init.maxInboundStreams ?? MAX_INBOUND_STREAMS
-    this.maxOutboundStreams = init.maxOutboundStreams ?? MAX_OUTBOUND_STREAMS
-    this.runOnTransientConnection = init.runOnTransientConnection ?? RUN_ON_TRANSIENT_CONNECTION
   }
 
   async start (): Promise<void> {

packages/transport-circuit-relay-v2/src/server/advert-service.ts

@@ -1,5 +1,6 @@
 import { TypedEventEmitter } from '@libp2p/interface/events'
 import pRetry from 'p-retry'
+import { number, object } from 'yup'
 import {
   DEFAULT_ADVERT_BOOT_DELAY,
   ERR_NO_ROUTERS_AVAILABLE,
@@ -30,6 +31,10 @@ export interface AdvertServiceEvents {
   'advert:error': CustomEvent<Error>
 }
 
+const configValidator = object({
+  bootDelay: number().integer().min(0).default(DEFAULT_ADVERT_BOOT_DELAY)
+})
+
 export class AdvertService extends TypedEventEmitter<AdvertServiceEvents> implements Startable {
   private readonly contentRouting: ContentRouting
   private timeout?: any
@@ -45,8 +50,11 @@ export class AdvertService extends TypedEventEmitter<AdvertServiceEvents> implem
 
     this.log = components.logger.forComponent('libp2p:circuit-relay:advert-service')
     this.contentRouting = components.contentRouting
-    this.bootDelay = init?.bootDelay ?? DEFAULT_ADVERT_BOOT_DELAY
     this.started = false
+
+    const config = configValidator.validateSync(init)
+
+    this.bootDelay = config.bootDelay
   }
 
   isStarted (): boolean {

packages/transport-circuit-relay-v2/src/transport/reservation-store.ts

@@ -4,6 +4,7 @@ import { PeerJobQueue } from '@libp2p/utils/peer-job-queue'
 import { multiaddr } from '@multiformats/multiaddr'
 import { pbStream } from 'it-protobuf-stream'
 import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
+import { number, object } from 'yup'
 import { DEFAULT_RESERVATION_CONCURRENCY, RELAY_TAG, RELAY_V2_HOP_CODEC } from '../constants.js'
 import { HopMessage, Status } from '../pb/index.js'
 import { getExpirationMilliseconds } from '../utils.js'
@@ -75,6 +76,13 @@ export interface ReservationStoreEvents {
   'relay:removed': CustomEvent<PeerId>
 }
 
+const configValidator = object({
+  discoverRelays: number().integer().min(0).default(0),
+  maxReservationQueueLength: number().integer().min(0).default(100),
+  reservationCompletionTimeout: number().integer().min(0).default(10000),
+  reservationConcurrency: number().integer().min(0).default(DEFAULT_RESERVATION_CONCURRENCY)
+})
+
 export class ReservationStore extends TypedEventEmitter<ReservationStoreEvents> implements Startable {
   private readonly peerId: PeerId
   private readonly connectionManager: ConnectionManager
@@ -99,14 +107,17 @@ export class ReservationStore extends TypedEventEmitter<ReservationStoreEvents>
     this.peerStore = components.peerStore
     this.events = components.events
     this.reservations = new PeerMap()
-    this.maxDiscoveredRelays = init?.discoverRelays ?? 0
-    this.maxReservationQueueLength = init?.maxReservationQueueLength ?? 100
-    this.reservationCompletionTimeout = init?.reservationCompletionTimeout ?? 10000
+
+    const config = configValidator.validateSync(init)
+
+    this.maxDiscoveredRelays = config.discoverRelays
+    this.maxReservationQueueLength = config.maxReservationQueueLength
+    this.reservationCompletionTimeout = config.reservationCompletionTimeout
     this.started = false
 
     // ensure we don't listen on multiple relays simultaneously
     this.reserveQueue = new PeerJobQueue({
-      concurrency: init?.reservationConcurrency ?? DEFAULT_RESERVATION_CONCURRENCY
+      concurrency: config.reservationConcurrency
     })
 
     // When a peer disconnects, if we had a reservation on that peer