feat: BlockBroker factory support

packages/helia/src/block-brokers/bitswap-block-broker.ts

@@ -1,4 +1,5 @@
 import { createBitswap } from 'ipfs-bitswap'
+import type { BlockBrokerFactoryFunction } from '@helia/interface'
 import type { BlockAnnouncer, BlockRetriever } from '@helia/interface/blocks'
 import type { Libp2p } from '@libp2p/interface'
 import type { Startable } from '@libp2p/interface/startable'
@@ -56,3 +57,11 @@ ProgressOptions<BitswapWantBlockProgressEvents>
     return this.bitswap.want(cid, options)
   }
 }
+
+/**
+ * A helper factory for users who want to override Helia `blockBrokers` but
+ * still want to use the default `BitswapBlockBroker`.
+ */
+export const BitswapBlockBrokerFactory: BlockBrokerFactoryFunction = (components): BitswapBlockBroker => {
+  return new BitswapBlockBroker(components.libp2p, components.blockstore, components.hashers)
+}

packages/helia/src/block-brokers/trustless-gateway-block-broker.ts

@@ -7,9 +7,10 @@ import type { ProgressEvent, ProgressOptions } from 'progress-events'
 const log = logger('helia:trustless-gateway-block-broker')
 
 /**
- * A BlockProvider constructs instances of `TrustlessGateway`
- * keeps track of the number of attempts, errors, and successes for a given
- * gateway url.
+ * A `TrustlessGateway` keeps track of the number of attempts, errors, and
+ * successes for a given gateway url so that we can prioritize gateways that
+ * have been more reliable in the past, and ensure that requests are distributed
+ * across all gateways within a given `TrustedGatewayBlockBroker` instance.
  */
 class TrustlessGateway {
   public readonly url: URL
@@ -24,14 +25,16 @@ class TrustlessGateway {
   /**
    * The number of times this gateway has errored while attempting to fetch a
    * block. This includes `response.ok === false` and any other errors that
-   * throw while attempting to fetch a block.
+   * throw while attempting to fetch a block. This does not include aborted
+   * attempts.
    */
   #errors = 0
 
   /**
    * The number of times this gateway has successfully fetched a block.
    */
   #successes = 0
+
   constructor (url: URL | string) {
     this.url = url instanceof URL ? url : new URL(url)
   }
@@ -85,22 +88,26 @@ class TrustlessGateway {
    * reliable, for prioritization. This is based on the number of successful attempts made
    * and the number of errors encountered.
    *
-   * * Unused gateways have 100% reliability
-   * * Gateways that have never errored have 100% reliability
+   * Unused gateways have 100% reliability; They will be prioritized over
+   * gateways with a 100% success rate to ensure that we attempt all gateways.
    */
   get reliability (): number {
-    // if we have never tried to use this gateway, it is considered the most
-    // reliable until we determine otherwise
-    // (prioritize unused gateways)
+    /**
+     * if we have never tried to use this gateway, it is considered the most
+     * reliable until we determine otherwise (prioritize unused gateways)
+     */
     if (this.#attempts === 0) {
       return 1
     }
 
-    // We have attempted the gateway, so we need to calculate the reliability
-    // based on the number of attempts, errors, and successes. Gateways that
-    // return a single error should drop their reliability score more than a
-    // success increases it.
-    // Play around with the below reliability function at https://www.desmos.com/calculator/d6hfhf5ukm
+    /**
+     * We have attempted the gateway, so we need to calculate the reliability
+     * based on the number of attempts, errors, and successes. Gateways that
+     * return a single error should drop their reliability score more than a
+     * single success increases it.
+     *
+     * Play around with the below reliability function at https://www.desmos.com/calculator/d6hfhf5ukm
+     */
     return this.#successes / (this.#attempts + (this.#errors * 3))
   }
 }

packages/helia/src/helia.ts

@@ -1,3 +1,4 @@
+import { type BlockBroker } from '@helia/interface/blocks'
 import { start, stop } from '@libp2p/interface/startable'
 import { logger } from '@libp2p/logger'
 import drain from 'it-drain'
@@ -19,6 +20,7 @@ const log = logger('helia')
 interface HeliaImplInit<T extends Libp2p = Libp2p> extends HeliaInit<T> {
   libp2p: T
   blockstore: Blockstore
+  blockBrokers: BlockBroker[]
   datastore: Datastore
 }
 

packages/helia/src/index.ts

@@ -30,7 +30,7 @@ import { defaultHashers } from './utils/default-hashers.js'
 import { createLibp2p } from './utils/libp2p.js'
 import { name, version } from './version.js'
 import type { DefaultLibp2pServices } from './utils/libp2p-defaults.js'
-import type { Helia } from '@helia/interface'
+import type { Helia, BlockBrokerFactoryFunction } from '@helia/interface'
 import type { BlockBroker } from '@helia/interface/blocks'
 import type { Libp2p } from '@libp2p/interface'
 import type { Blockstore } from 'interface-blockstore'
@@ -98,7 +98,7 @@ export interface HeliaInit<T extends Libp2p = Libp2p> {
    * A list of strategies used to fetch blocks when they are not present in
    * the local blockstore
    */
-  blockBrokers?: BlockBroker[]
+  blockBrokers?: Array<BlockBroker | BlockBrokerFactoryFunction>
 
   /**
    * Pass `false` to not start the Helia node
@@ -159,7 +159,17 @@ export async function createHelia (init: HeliaInit = {}): Promise<Helia<unknown>
 
   const hashers = defaultHashers(init.hashers)
 
-  const blockBrokers = init.blockBrokers ?? [
+  const blockBrokers: BlockBroker[] = init.blockBrokers?.map((blockBroker: BlockBroker | BlockBrokerFactoryFunction): BlockBroker => {
+    if (typeof blockBroker !== 'function') {
+      return blockBroker satisfies BlockBroker
+    }
+    return blockBroker({
+      blockstore,
+      datastore,
+      libp2p,
+      hashers
+    }) satisfies BlockBroker
+  }) ?? [
     new BitswapBlockBroker(libp2p, blockstore, hashers),
     new TrustedGatewayBlockBroker(DEFAULT_TRUSTLESS_GATEWAYS)
   ]

packages/helia/test/fixtures/create-helia.ts

@@ -2,11 +2,15 @@ import { webSockets } from '@libp2p/websockets'
 import * as Filters from '@libp2p/websockets/filters'
 import { circuitRelayTransport } from 'libp2p/circuit-relay'
 import { identifyService } from 'libp2p/identify'
+import { BitswapBlockBrokerFactory } from '../../src/block-brokers/bitswap-block-broker.js'
 import { createHelia as createNode } from '../../src/index.js'
 import type { Helia } from '@helia/interface'
 
 export async function createHelia (): Promise<Helia> {
   return createNode({
+    blockBrokers: [
+      BitswapBlockBrokerFactory
+    ],
     libp2p: {
       addresses: {
         listen: [

packages/helia/test/pins.depth-limited.spec.ts

@@ -27,6 +27,7 @@ describe('pins (depth limited)', () => {
     dag = await createDag(codec, blockstore, MAX_DEPTH, 3)
 
     helia = await createHelia({
+      blockBrokers: [],
       datastore: new MemoryDatastore(),
       blockstore,
       libp2p: await createLibp2p({

packages/helia/test/pins.recursive.spec.ts

@@ -25,6 +25,7 @@ describe('pins (recursive)', () => {
     dag = await createDag(codec, blockstore, 2, 3)
 
     helia = await createHelia({
+      blockBrokers: [],
       datastore: new MemoryDatastore(),
       blockstore,
       libp2p: await createLibp2p({

packages/helia/test/utils/networked-storage.spec.ts

@@ -15,7 +15,7 @@ import type { BitswapBlockBroker } from '../../src/block-brokers/bitswap-block-b
 import type { Blockstore } from 'interface-blockstore'
 import type { CID } from 'multiformats/cid'
 
-describe('storage', () => {
+describe('networked-storage', () => {
   let storage: NetworkedStorage
   let blockstore: Blockstore
   let bitswap: StubbedInstance<BitswapBlockBroker>

packages/interface/src/index.ts

@@ -14,11 +14,12 @@
  * ```
  */
 
-import type { Blocks } from './blocks.js'
+import type { BlockBroker, Blocks } from './blocks.js'
 import type { Pins } from './pins.js'
 import type { Libp2p, AbortOptions } from '@libp2p/interface'
 import type { Datastore } from 'interface-datastore'
 import type { CID } from 'multiformats/cid'
+import type { MultihashHasher } from 'multiformats/hashes/interface'
 import type { ProgressEvent, ProgressOptions } from 'progress-events'
 
 export type { Await, AwaitIterable } from 'interface-store'
@@ -70,3 +71,18 @@ export type GcEvents =
 export interface GCOptions extends AbortOptions, ProgressOptions<GcEvents> {
 
 }
+export type BlockBrokerFactoryComponents = Pick<Helia, 'libp2p' | 'blockstore' | 'datastore'> & {
+  hashers: MultihashHasher[]
+}
+
+/**
+ * A function that receives some {@link Helia} components and returns a
+ * {@link BlockBroker}.
+ *
+ * This is needed in order to re-use some of the internal components Helia
+ * constructs without having to hoist each required component into the top-level
+ * scope.
+ */
+export interface BlockBrokerFactoryFunction {
+  (heliaComponents: BlockBrokerFactoryComponents): BlockBroker
+}