diff --git a/packages/fallback-router/src/index.ts b/packages/fallback-router/src/index.ts index 026d3a0b6..d46876c78 100644 --- a/packages/fallback-router/src/index.ts +++ b/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, options?: RoutingOptions | undefined): AsyncIterable { yield * (this.shuffle ? this.gateways.toSorted(() => Math.random() > 0.5 ? 1 : -1) diff --git a/packages/helia/src/routing.ts b/packages/helia/src/routing.ts index 7b1c290f3..a9f2f1305 100644 --- a/packages/helia/src/routing.ts +++ b/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({ - 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>[]): AsyncGenerator { + 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({ + 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 (routers: any[], key: Operation): Array> { +function supports (routers: any[], key: Operation): Array> & Pick> { return routers.filter(router => router[key] != null) } diff --git a/packages/helia/test/routing.spec.ts b/packages/helia/test/routing.spec.ts index e1b1c9abd..e79fa9f03 100644 --- a/packages/helia/test/routing.spec.ts +++ b/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 beforeEach(() => { - routerA = stubInterface() - routerB = stubInterface() + routerA = stubInterface({ + name: 'routerA' + }) + routerB = stubInterface({ + 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({ + multiaddrs: [stubInterface()] + }) + })()) + + 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') + }) }) diff --git a/packages/interface/src/routing.ts b/packages/interface/src/routing.ts index 68a74ab3e..765568a43 100644 --- a/packages/interface/src/routing.ts +++ b/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): AsyncIterable } -export type Routing = Required> +export type Routing = Required>