feat: emit standardized block broker events

So callers do not kneed to know how a node is configured in advance,
emit standardized events for block broker operations.

Author: achingbrain

packages/bitswap/src/bitswap.ts

@@ -7,7 +7,7 @@ import { createBitswapSession } from './session.ts'
 import { Stats } from './stats.ts'
 import { WantList } from './want-list.ts'
 import type { BitswapOptions, Bitswap as BitswapInterface, BitswapWantProgressEvents, BitswapNotifyProgressEvents, WantListEntry, BitswapComponents, PeerWantListEntry } from './index.ts'
-import type { CreateSessionOptions, ProviderOptions, SessionBlockBroker } from '@helia/interface'
+import type { BlockBrokerReceiveBlockProgressEvent, CreateSessionOptions, ProviderOptions, SessionBlockBroker } from '@helia/interface'
 import type { ComponentLogger, Libp2p, PeerId, AbortOptions } from '@libp2p/interface'
 import type { Logger } from '@libp2p/logger'
 import type { Blockstore } from 'interface-blockstore'
@@ -96,7 +96,7 @@ export class Bitswap implements BitswapInterface {
         signal
       })
 
-      options.onProgress?.(new CustomProgressEvent<{ cid: CID, sender: PeerId }>('bitswap:block', { cid, sender: result.sender }))
+      options.onProgress?.(new CustomProgressEvent<{ cid: CID, sender: PeerId }>('bitswap:block', { cid, sender: result.connection.remotePeer }))
 
       return result.block
     } finally {

packages/bitswap/src/constants.ts

@@ -2,9 +2,9 @@ export const BITSWAP_120 = '/ipfs/bitswap/1.2.0'
 export const DEFAULT_MAX_SIZE_REPLACE_HAS_WITH_BLOCK = 1024
 export const DEFAULT_MAX_INBOUND_STREAMS = 1024
 export const DEFAULT_MAX_OUTBOUND_STREAMS = 1024
-export const DEFAULT_MESSAGE_RECEIVE_TIMEOUT = 5000
-export const DEFAULT_MESSAGE_SEND_DELAY = 10
-export const DEFAULT_MESSAGE_SEND_TIMEOUT = 5000
+export const DEFAULT_MESSAGE_RECEIVE_TIMEOUT = 10_000
+export const DEFAULT_WANTLIST_SEND_DEBOUNCE = 10
+export const DEFAULT_MESSAGE_SEND_TIMEOUT = 10_000
 export const DEFAULT_MESSAGE_SEND_CONCURRENCY = 50
 export const DEFAULT_RUN_ON_TRANSIENT_CONNECTIONS = false
 export const DEFAULT_SESSION_ROOT_PRIORITY = 1

packages/bitswap/src/index.ts

@@ -9,7 +9,7 @@
 import { Bitswap as BitswapClass } from './bitswap.ts'
 import type { BitswapNetworkNotifyProgressEvents, BitswapNetworkWantProgressEvents, BitswapNetworkProgressEvents } from './network.ts'
 import type { WantType } from './pb/message.ts'
-import type { CreateSessionOptions, ProviderOptions, SessionBlockBroker } from '@helia/interface'
+import type { Await, BlockBrokerGetBlockProgressEvents, CreateSessionOptions, ProviderOptions, SessionBlockBroker } from '@helia/interface'
 import type { Routing } from '@helia/interface/routing'
 import type { Libp2p, AbortOptions, Startable, ComponentLogger, Metrics, PeerId } from '@libp2p/interface'
 import type { Blockstore } from 'interface-blockstore'
@@ -27,7 +27,8 @@ export type BitswapWantBlockProgressEvents =
   ProgressEvent<'bitswap:unwant', CID> |
   ProgressEvent<'bitswap:want', CID> |
   ProgressEvent<'bitswap:block', { cid: CID, sender: PeerId }> |
-  BitswapNetworkWantProgressEvents
+  BitswapNetworkWantProgressEvents |
+  BlockBrokerGetBlockProgressEvents
 
 export type { BitswapNetworkNotifyProgressEvents }
 export type { BitswapNetworkWantProgressEvents }
@@ -76,7 +77,7 @@ export interface Bitswap extends Startable {
 }
 
 export interface MultihashHasherLoader {
-  getHasher(codeOrName: number | string): Promise<MultihashHasher>
+  getHasher(codeOrName: number | string): Await<MultihashHasher>
 }
 
 export interface BitswapComponents {
@@ -85,6 +86,7 @@ export interface BitswapComponents {
   logger: ComponentLogger
   libp2p: Libp2p
   metrics?: Metrics
+  hashLoader?: MultihashHasherLoader
 }
 
 export interface BitswapOptions {
@@ -118,11 +120,6 @@ export interface BitswapOptions {
    */
   runOnLimitedConnections?: boolean
 
-  /**
-   * Enables loading esoteric hash functions
-   */
-  hashLoader?: MultihashHasherLoader
-
   /**
    * The protocol that we speak
    *
@@ -137,6 +134,21 @@ export interface BitswapOptions {
    */
   messageSendConcurrency?: number
 
+  /**
+   * Wantlist or blocks must be sent to peers within this many milliseconds
+   *
+   * @default 10_000
+   */
+  messageSendTimeout?: number
+
+  /**
+   * Incoming wantlist or block messages must be received from peers within this
+   * many milliseconds
+   *
+   * @default 10_000
+   */
+  messageReceiveTimeout?: number
+
   /**
    * When sending blocks to peers, how many messages to send at once
    *
@@ -149,7 +161,7 @@ export interface BitswapOptions {
    * This is useful for preventing slow/large peer-connections from consuming
    * your bandwidth/streams.
    *
-   * @default 10000
+   * @default 10_000
    */
   sendBlocksTimeout?: number
 
@@ -162,6 +174,15 @@ export interface BitswapOptions {
    */
   sendBlocksDebounce?: number
 
+  /**
+   * When a want is added to the local wantlist, wait this many milliseconds
+   * before sending the wantlist to connected or session peers in case more
+   * wants are about to be added
+   *
+   * @default 10
+   */
+  sendWantlistDebounce?: number
+
   /**
    * If the client sends a want-have, and we have the corresponding block, we
    * check the size of the block and if it's small enough we send the block

packages/bitswap/src/network.ts

@@ -7,7 +7,7 @@ import { pushable } from 'it-pushable'
 import take from 'it-take'
 import { CustomProgressEvent } from 'progress-events'
 import { raceEvent } from 'race-event'
-import { BITSWAP_120, DEFAULT_MAX_INBOUND_STREAMS, DEFAULT_MAX_INCOMING_MESSAGE_SIZE, DEFAULT_MAX_OUTBOUND_STREAMS, DEFAULT_MAX_OUTGOING_MESSAGE_SIZE, DEFAULT_MAX_PROVIDERS_PER_REQUEST, DEFAULT_MESSAGE_RECEIVE_TIMEOUT, DEFAULT_MESSAGE_SEND_CONCURRENCY, DEFAULT_RUN_ON_TRANSIENT_CONNECTIONS } from './constants.ts'
+import { BITSWAP_120, DEFAULT_MAX_INBOUND_STREAMS, DEFAULT_MAX_INCOMING_MESSAGE_SIZE, DEFAULT_MAX_OUTBOUND_STREAMS, DEFAULT_MAX_OUTGOING_MESSAGE_SIZE, DEFAULT_MAX_PROVIDERS_PER_REQUEST, DEFAULT_MESSAGE_RECEIVE_TIMEOUT, DEFAULT_MESSAGE_SEND_CONCURRENCY, DEFAULT_MESSAGE_SEND_TIMEOUT, DEFAULT_RUN_ON_TRANSIENT_CONNECTIONS } from './constants.ts'
 import { BitswapMessage } from './pb/message.ts'
 import { mergeMessages } from './utils/merge-messages.ts'
 import { splitMessage } from './utils/split-message.ts'
@@ -16,13 +16,14 @@ import type { MultihashHasherLoader } from './index.ts'
 import type { Block } from './pb/message.ts'
 import type { QueuedBitswapMessage } from './utils/bitswap-message.ts'
 import type { Provider, Routing, RoutingFindProvidersProgressEvents } from '@helia/interface/routing'
-import type { Libp2p, AbortOptions, Connection, PeerId, Topology, ComponentLogger, IdentifyResult, Counter, Metrics, Stream } from '@libp2p/interface'
+import type { Libp2p, AbortOptions, Connection, PeerId, Topology, ComponentLogger, IdentifyResult, Counter, Metrics, Stream, OpenConnectionProgressEvents, NewStreamProgressEvents } from '@libp2p/interface'
 import type { Logger } from '@libp2p/logger'
 import type { PeerQueueJobOptions } from '@libp2p/utils'
 import type { Multiaddr } from '@multiformats/multiaddr'
-import type { CID } from 'multiformats/cid'
+import { CID } from 'multiformats/cid'
 import type { ProgressEvent, ProgressOptions } from 'progress-events'
 import type { Uint8ArrayList } from 'uint8arraylist'
+import type { BlockBrokerGetBlockProgressEvents } from '@helia/interface'
 
 export interface BitswapProvider {
   /**
@@ -47,15 +48,18 @@ export interface BitswapProvider {
 }
 
 export type BitswapNetworkProgressEvents =
-  ProgressEvent<'bitswap:dial', PeerId | Multiaddr | Multiaddr[]>
+  ProgressEvent<'bitswap:dial', PeerId | Multiaddr | Multiaddr[]> |
+  OpenConnectionProgressEvents |
+  NewStreamProgressEvents
 
 export type BitswapNetworkWantProgressEvents =
   ProgressEvent<'bitswap:send-wantlist', PeerId> |
   ProgressEvent<'bitswap:send-wantlist:error', { peer: PeerId, error: Error }> |
   ProgressEvent<'bitswap:find-providers', CID> |
   ProgressEvent<'bitswap:found-provider', BitswapProvider> |
   BitswapNetworkProgressEvents |
-  RoutingFindProvidersProgressEvents
+  RoutingFindProvidersProgressEvents |
+  BlockBrokerGetBlockProgressEvents
 
 export type BitswapNetworkNotifyProgressEvents =
   BitswapNetworkProgressEvents |
@@ -66,6 +70,7 @@ export interface NetworkInit {
   maxInboundStreams?: number
   maxOutboundStreams?: number
   messageReceiveTimeout?: number
+  messageSendTimeout?: number
   messageSendConcurrency?: number
   protocols?: string[]
   runOnLimitedConnections?: boolean
@@ -81,12 +86,16 @@ export interface NetworkComponents {
 }
 
 export interface BitswapMessageEventDetail {
+  /**
+   * @deprecated access the peer via connection.remotePeer instead
+   */
   peer: PeerId
   message: BitswapMessage
+  connection: Connection
 }
 
 export interface NetworkEvents {
-  'bitswap:message': CustomEvent<{ peer: PeerId, message: BitswapMessage }>
+  'bitswap:message': CustomEvent<BitswapMessageEventDetail>
   'peer:connected': CustomEvent<PeerId>
   'peer:disconnected': CustomEvent<PeerId>
 }
@@ -104,6 +113,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
   private readonly maxInboundStreams: number
   private readonly maxOutboundStreams: number
   private readonly messageReceiveTimeout: number
+  private readonly messageSendTimeout: number
   private registrarIds: string[]
   private readonly metrics: { blocksSent?: Counter, dataSent?: Counter }
   private readonly sendQueue: PeerQueue<void, SendMessageJobOptions>
@@ -126,6 +136,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
     this.maxInboundStreams = init.maxInboundStreams ?? DEFAULT_MAX_INBOUND_STREAMS
     this.maxOutboundStreams = init.maxOutboundStreams ?? DEFAULT_MAX_OUTBOUND_STREAMS
     this.messageReceiveTimeout = init.messageReceiveTimeout ?? DEFAULT_MESSAGE_RECEIVE_TIMEOUT
+    this.messageSendTimeout = init.messageSendTimeout ?? DEFAULT_MESSAGE_SEND_TIMEOUT
     this.runOnLimitedConnections = init.runOnLimitedConnections ?? DEFAULT_RUN_ON_TRANSIENT_CONNECTIONS
     this.maxIncomingMessageSize = init.maxIncomingMessageSize ?? DEFAULT_MAX_OUTGOING_MESSAGE_SIZE
     this.maxOutgoingMessageSize = init.maxOutgoingMessageSize ?? init.maxIncomingMessageSize ?? DEFAULT_MAX_INCOMING_MESSAGE_SIZE
@@ -245,10 +256,11 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
           const message = BitswapMessage.decode(data)
           this.log('incoming new bitswap %s message from %p on stream', stream.protocol, connection.remotePeer, stream.id)
 
-          this.safeDispatchEvent('bitswap:message', {
+          this.safeDispatchEvent<BitswapMessageEventDetail>('bitswap:message', {
             detail: {
               peer: connection.remotePeer,
-              message
+              message,
+              connection
             }
           })
 
@@ -287,7 +299,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
         continue
       }
 
-      options?.onProgress?.(new CustomProgressEvent('bitswap:found-provider', {
+      options?.onProgress?.(new CustomProgressEvent<BitswapProvider>('bitswap:found-provider', {
         type: 'bitswap',
         cid,
         provider,
@@ -325,8 +337,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
   }
 
   /**
-   * Connect to the given peer
-   * Send the given msg (instance of Message) to the given peer
+   * Connect to the specified peer and send the given message
    */
   async sendMessage (peerId: PeerId, message: QueuedBitswapMessage, options?: AbortOptions & ProgressOptions<BitswapNetworkWantProgressEvents>): Promise<void> {
     if (!this.running) {
@@ -359,7 +370,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
       options?.onProgress?.(new CustomProgressEvent<PeerId>('bitswap:network:send-wantlist', peerId))
 
       const stream = await this.libp2p.dialProtocol(peerId, BITSWAP_120, options)
-      await stream.closeRead()
+      await stream.closeRead(options)
 
       try {
         for (const buf of splitMessage(message, this.maxOutgoingMessageSize)) {
@@ -377,8 +388,9 @@ export class Network extends TypedEventEmitter<NetworkEvents> {
 
       this._updateSentStats(message.blocks)
     }, {
+      onProgress: options?.onProgress,
       peerId,
-      signal: options?.signal,
+      signal: options?.signal ?? AbortSignal.timeout(this.messageSendTimeout),
       message
     })
   }

packages/bitswap/src/peer-want-lists/index.ts

@@ -56,9 +56,9 @@ export class PeerWantLists {
     })
 
     this.network.addEventListener('bitswap:message', (evt) => {
-      this.receiveMessage(evt.detail.peer, evt.detail.message)
+      this.receiveMessage(evt.detail.connection.remotePeer, evt.detail.message)
         .catch(err => {
-          this.log.error('error receiving bitswap message from %p - %e', evt.detail.peer, err)
+          this.log.error('error receiving bitswap message from %p - %e', evt.detail.connection.remotePeer, err)
         })
     })
     this.network.addEventListener('peer:disconnected', evt => {