feat: add router capabilities for fallback routers

packages/fallback-router/src/index.ts

@@ -14,7 +14,7 @@ import { CID } from 'multiformats/cid'
 import { identity } from 'multiformats/hashes/identity'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
-import type { Peer, Provider, Router, RoutingOptions } from '@helia/interface'
+import type { Capability, Peer, Provider, Router, RoutingOptions } from '@helia/interface'
 import type { Version } from 'multiformats'
 
 export interface FallbackRouterInit {
@@ -60,6 +60,12 @@ class FallbackRouter implements Router {
     this.shuffle = init.shuffle ?? true
   }
 
+  capabilities (): Capability[] {
+    return [
+      'fallback'
+    ]
+  }
+
   async * findProviders (cid: CID<unknown, number, number, Version>, options?: RoutingOptions | undefined): AsyncIterable<Provider> {
     yield * (this.shuffle
       ? this.gateways.toSorted(() => Math.random() > 0.5 ? 1 : -1)

packages/helia/src/routing.ts

@@ -88,23 +88,14 @@ export class Routing implements RoutingInterface, Startable {
       throw new NoRoutersAvailableError('No content routers available')
     }
 
-    // provider multiaddrs are only cached for a limited time, so they can come
-    // back as an empty array - when this happens we have to do a FIND_PEER
-    // query to get updated addresses, but we shouldn't block on this so use a
-    // separate bounded queue to perform this lookup
-    const queue = new Queue<Provider | null, PeerQueueOptions>({
-      concurrency: this.providerLookupConcurrency
-    })
-
     let foundProviders = 0
     const errors: Error[] = []
     const self = this
-    let routersFinished = 0
 
-    this.log('findProviders for %c start using routers %s', key, this.routers.map(r => r.toString()).join(', '))
+    async function * findProviders (routers: Required<Pick<Router, 'findProviders' | 'name'>>[]): AsyncGenerator<Provider> {
+      let routersFinished = 0
 
-    const routers = supports(this.routers, 'findProviders')
-      .map(async function * (router) {
+      const streams = routers.map(async function * (router) {
         let foundProviders = 0
 
         options?.onProgress?.(new CustomProgressEvent('helia:routing:find-providers:start', {
@@ -128,7 +119,7 @@ export class Routing implements RoutingInterface, Startable {
         } catch (err: any) {
           errors.push(err)
         } finally {
-          self.log('router %s found %d providers for %c', router, foundProviders, key)
+          self.log('router %s found %d providers for %c', router.name, foundProviders, key)
 
           options?.onProgress?.(new CustomProgressEvent('helia:routing:find-providers:end', {
             router: router.name,
@@ -146,53 +137,77 @@ export class Routing implements RoutingInterface, Startable {
         }
       })
 
-    for await (const peer of merge(
-      queue.toGenerator(),
-      ...routers)
-    ) {
-      // the peer was yielded by a content router without multiaddrs and we
-      // failed to load them
-      if (peer == null) {
-        continue
-      }
+      // provider multiaddrs are only cached for a limited time, so they can come
+      // back as an empty array - when this happens we have to do a FIND_PEER
+      // query to get updated addresses, but we shouldn't block on this so use a
+      // separate bounded queue to perform this lookup
+      const queue = new Queue<Provider | null, PeerQueueOptions>({
+        concurrency: self.providerLookupConcurrency
+      })
 
-      // have to refresh peer info for this peer to get updated multiaddrs
-      if (peer.multiaddrs.length === 0) {
-        // already looking this peer up
-        if (queue.queue.find(job => job.options.peer.equals(peer.id)) != null) {
+      for await (const peer of merge(
+        queue.toGenerator(),
+        ...streams)
+      ) {
+        // the peer was yielded by a content router without multiaddrs and we
+        // failed to load them
+        if (peer == null) {
           continue
         }
 
-        queue.add(async () => {
-          try {
-            const provider = await this.findPeer(peer.id, options)
+        // have to refresh peer info for this peer to get updated multiaddrs
+        if (peer.multiaddrs.length === 0) {
+          // already looking this peer up
+          if (queue.queue.find(job => job.options.peer.equals(peer.id)) != null) {
+            continue
+          }
 
-            if (provider.multiaddrs.length === 0) {
+          queue.add(async () => {
+            try {
+              const provider = await self.findPeer(peer.id, options)
+
+              if (provider.multiaddrs.length === 0) {
+                return null
+              }
+
+              return {
+                ...provider,
+                protocols: peer.protocols,
+                router: peer.router
+              }
+            } catch (err) {
+              self.log.error('could not load multiaddrs for peer %p - %e', peer.id, err)
               return null
             }
-
-            return {
-              ...provider,
-              protocols: peer.protocols,
-              router: peer.router
-            }
-          } catch (err) {
-            this.log.error('could not load multiaddrs for peer %p - %e', peer.id, err)
-            return null
-          }
-        }, {
-          peer: peer.id,
-          signal: options.signal
-        })
-          .catch(err => {
-            this.log.error('could not load multiaddrs for peer %p - %e', peer.id, err)
+          }, {
+            peer: peer.id,
+            signal: options.signal
           })
+            .catch(err => {
+              self.log.error('could not load multiaddrs for peer %p - %e', peer.id, err)
+            })
 
-        continue
+          continue
+        }
+
+        foundProviders++
+        yield peer
       }
+    }
 
-      foundProviders++
-      yield peer
+    const routers = supports(this.routers, 'findProviders')
+    const defaultRouters = routers.filter(r => r.capabilities == null || !r.capabilities().includes('fallback'))
+    const fallbackRouters = routers.filter(r => r.capabilities?.().includes('fallback') === true)
+
+    this.log('findProviders for %c start using routers %s', key, defaultRouters.map(r => r.name).join(', '))
+
+    // use non-fallback routers first
+    yield * findProviders(defaultRouters)
+
+    // only use fallback routers if no providers have been found
+    if (foundProviders === 0 && fallbackRouters.length > 0) {
+      this.log('findProviders for %c using fallback routers %s', key, fallbackRouters.map(r => r.name).join(', '))
+      yield * findProviders(fallbackRouters)
     }
 
     this.log('findProviders finished, found %d providers for %c', foundProviders, key)
@@ -296,7 +311,7 @@ export class Routing implements RoutingInterface, Startable {
                 signal
               })
             } catch (err: any) {
-              this.log('router %s failed with %e', router, err)
+              this.log('router %s failed with %e', router.name, err)
               errors.push(err)
               throw err
             } finally {
@@ -412,6 +427,6 @@ export class Routing implements RoutingInterface, Startable {
   }
 }
 
-function supports <Operation extends keyof Routing> (routers: any[], key: Operation): Array<Pick<Routing, Operation | 'name'>> {
+function supports <Operation extends keyof Router> (routers: any[], key: Operation): Array<Required<Pick<Router, Operation | 'name'>> & Pick<Router, 'capabilities'>> {
   return routers.filter(router => router[key] != null)
 }

packages/helia/test/routing.spec.ts

@@ -4,7 +4,8 @@ import all from 'it-all'
 import { CID } from 'multiformats/cid'
 import { stubInterface } from 'sinon-ts'
 import { Routing } from '../src/routing.ts'
-import type { Router } from '@helia/interface'
+import type { Provider, Router } from '@helia/interface'
+import type { Multiaddr } from '@multiformats/multiaddr'
 import type { StubbedInstance } from 'sinon-ts'
 
 describe('routing', () => {
@@ -13,8 +14,15 @@ describe('routing', () => {
   let routerB: StubbedInstance<Router>
 
   beforeEach(() => {
-    routerA = stubInterface<Router>()
-    routerB = stubInterface<Router>()
+    routerA = stubInterface<Router>({
+      name: 'routerA'
+    })
+    routerB = stubInterface<Router>({
+      name: 'routerB'
+    })
+
+    routerA.capabilities?.returns([])
+    routerB.capabilities?.returns([])
 
     routing = new Routing({
       logger: defaultLogger()
@@ -34,4 +42,67 @@ describe('routing', () => {
 
     await expect(all(routing.findProviders(key))).to.eventually.be.empty()
   })
+
+  it('should call routers in declaration order', async () => {
+    const key = CID.parse('QmaQwYWpchozXhFv8nvxprECWBSCEppN9dfd2VQiJfRo3E')
+
+    let firstRouter: null | string = null
+
+    // eslint-disable-next-line require-yield
+    routerA.findProviders?.returns((async function * () {
+      if (firstRouter == null) {
+        firstRouter = 'a'
+      }
+    })())
+    // eslint-disable-next-line require-yield
+    routerB.findProviders?.returns((async function * () {
+      if (firstRouter == null) {
+        firstRouter = 'b'
+      }
+    })())
+
+    await expect(all(routing.findProviders(key))).to.eventually.be.empty()
+    expect(firstRouter).to.equal('a')
+  })
+
+  it('should use a fallback router after a regular router', async () => {
+    const key = CID.parse('QmaQwYWpchozXhFv8nvxprECWBSCEppN9dfd2VQiJfRo3E')
+
+    let firstRouter: null | string = null
+
+    // eslint-disable-next-line require-yield
+    routerA.findProviders?.returns((async function * () {
+      if (firstRouter == null) {
+        firstRouter = 'a'
+      }
+    })())
+    // eslint-disable-next-line require-yield
+    routerB.findProviders?.returns((async function * () {
+      if (firstRouter == null) {
+        firstRouter = 'b'
+      }
+    })())
+
+    routerA.capabilities?.returns(['fallback'])
+
+    await expect(all(routing.findProviders(key))).to.eventually.be.empty()
+    expect(firstRouter).to.equal('b')
+  })
+
+  it('should not use a fallback router if a regular router finds providers', async () => {
+    const key = CID.parse('QmaQwYWpchozXhFv8nvxprECWBSCEppN9dfd2VQiJfRo3E')
+
+    routerA.findProviders?.returns((async function * () {})())
+    routerB.findProviders?.returns((async function * () {
+      yield stubInterface<Provider>({
+        multiaddrs: [stubInterface<Multiaddr>()]
+      })
+    })())
+
+    routerA.capabilities?.returns(['fallback'])
+
+    await expect(all(routing.findProviders(key))).to.eventually.have.lengthOf(1, 'did not find provider')
+    expect(routerA.findProviders?.called).to.be.false('called fallback router')
+    expect(routerB.findProviders?.called).to.be.true('did not call regular router')
+  })
 })

packages/interface/src/routing.ts

@@ -219,12 +219,16 @@ export type RoutingGetClosestPeersProgressEvents =
   ProgressEvent<'helia:routing:get-closest-peers:start', RoutingGetClosestPeersStartEvent> |
   ProgressEvent<'helia:routing:get-closest-peers:end', RoutingGetClosestPeersEndEvent>
 
+export type Capability = 'fallback'
+
 export interface Router {
   /**
    * The name of this routing implementation
    */
   name: string
 
+  capabilities?(): Capability[]
+
   /**
    * The implementation of this method should ensure that network peers know the
    * caller can provide content that corresponds to the passed CID.
@@ -337,4 +341,4 @@ export interface Router {
   getClosestPeers?(key: Uint8Array, options?: RoutingOptions<RoutingGetClosestPeersProgressEvents>): AsyncIterable<Peer>
 }
 
-export type Routing = Required<Omit<Router, 'name'>>
+export type Routing = Required<Omit<Router, 'name' | 'capabilities'>>