From c2c4f7ee00a9b47b44ac9310be766dcd44497591 Mon Sep 17 00:00:00 2001 From: achingbrain Date: Fri, 10 Jul 2026 11:01:04 +0200 Subject: [PATCH] feat: add router capabilities for fallback routers Adds an optional `.capabilities` function to the router interface that can signal to Helia's routing system that the router is to be used after existing routers. The practical application is falling back to trustless gateways when delegated routing fails to find providers. --- packages/fallback-router/src/index.ts | 8 +- packages/helia/src/routing.ts | 117 +++++++++++++++----------- packages/helia/test/routing.spec.ts | 77 ++++++++++++++++- packages/interface/src/routing.ts | 6 +- 4 files changed, 152 insertions(+), 56 deletions(-) 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>