feat: Race top n nodes when making requests

src/client.js

@@ -12,6 +12,7 @@ import { isBrowserContext } from './utils/runtime.js'
 
 export class Saturn {
   static nodesListKey = 'saturn-nodes'
+  static defaultRaceCount = 3
   /**
    *
    * @param {object} [opts={}]
@@ -51,6 +52,101 @@ export class Saturn {
     this.loadNodesPromise = this._loadNodes(this.opts)
   }
 
+  /**
+   *
+   * @param {string} cidPath
+   * @param {object} [opts={}]
+   * @param {('car'|'raw')} [opts.format]
+   * @param {number} [opts.connectTimeout=5000]
+   * @param {number} [opts.downloadTimeout=0]
+   * @returns {Promise<object>}
+   */
+  async fetchCIDWithRace (cidPath, opts = {}) {
+    const [cid] = (cidPath ?? '').split('/')
+    CID.parse(cid)
+
+    const jwt = await getJWT(this.opts, this.storage)
+
+    const options = Object.assign({}, this.opts, { format: 'car', jwt }, opts)
+
+    if (!isBrowserContext) {
+      options.headers = {
+        ...(options.headers || {}),
+        Authorization: 'Bearer ' + options.jwt
+      }
+    }
+
+    const origins = options.origins
+    const controllers = []
+
+    const createFetchPromise = async (origin) => {
+      options.url = origin
+      const url = this.createRequestURL(cidPath, options)
+
+      const controller = new AbortController()
+      controllers.push(controller)
+      const connectTimeout = setTimeout(() => {
+        controller.abort()
+      }, options.connectTimeout)
+
+      try {
+        res = await fetch(parseUrl(url), { signal: controller.signal, ...options })
+        clearTimeout(connectTimeout)
+        return { res, url, controller }
+      } catch (err) {
+        throw new Error(
+          `Non OK response received: ${res.status} ${res.statusText}`
+        )
+      }
+    }
+
+    const abortRemainingFetches = async (res, controllers) => {
+      return controllers.forEach((controller) => {
+        if (res.controller !== controller) {
+          controller.abort('Request race unsuccessful')
+        }
+      })
+    }
+
+    const fetchPromises = Promise.any(origins.map((origin) => createFetchPromise(origin)))
+
+    const log = {
+      startTime: new Date()
+    }
+
+    let res, url, controller
+    try {
+      ({ res, url, controller } = await fetchPromises)
+
+      abortRemainingFetches(res, controllers)
+
+      const { headers } = res
+      log.url = url
+      log.ttfbMs = new Date() - log.startTime
+      log.httpStatusCode = res.status
+      log.cacheHit = headers.get('saturn-cache-status') === 'HIT'
+      log.nodeId = headers.get('saturn-node-id')
+      log.requestId = headers.get('saturn-transfer-id')
+      log.httpProtocol = headers.get('quic-status')
+
+      if (!res.ok) {
+        throw new Error(
+            `Non OK response received: ${res.status} ${res.statusText}`
+        )
+      }
+    } catch (err) {
+      if (!res) {
+        log.error = err.message
+      }
+      // Report now if error, otherwise report after download is done.
+      this._finalizeLog(log)
+
+      throw err
+    }
+
+    return { res, controller, log }
+  }
+
   /**
    *
    * @param {string} cidPath
@@ -68,7 +164,6 @@ export class Saturn {
 
     const options = Object.assign({}, this.opts, { format: 'car', jwt }, opts)
     const url = this.createRequestURL(cidPath, options)
-
     const log = {
       url,
       startTime: new Date()
@@ -122,6 +217,7 @@ export class Saturn {
    * @param {string} cidPath
    * @param {object} [opts={}]
    * @param {('car'|'raw')} [opts.format]
+   * @param {boolean} [opts.raceNodes]
    * @param {string} [opts.url]
    * @param {number} [opts.connectTimeout=5000]
    * @param {number} [opts.downloadTimeout=0]
@@ -166,11 +262,18 @@ export class Saturn {
     }
 
     let fallbackCount = 0
-    for (const origin of this.nodes) {
+    const nodes = this.nodes
+    for (let i = 0; i < nodes.length; i++) {
       if (fallbackCount > this.opts.fallbackLimit) {
         return
       }
-      opts.url = origin.url
+      if (opts.raceNodes) {
+        const origins = nodes.slice(i, i + Saturn.defaultRaceCount).map((node) => node.url)
+        opts.origins = origins
+      } else {
+        opts.url = nodes[i].url
+      }
+
       try {
         yield * fetchContent()
         return
@@ -189,13 +292,20 @@ export class Saturn {
    *
    * @param {string} cidPath
    * @param {object} [opts={}]
-   * @param {('car'|'raw')} [opts.format]
+   * @param {('car'|'raw')} [opts.format]- -
+   * @param {boolean} [opts.raceNodes]- -
    * @param {number} [opts.connectTimeout=5000]
    * @param {number} [opts.downloadTimeout=0]
    * @returns {Promise<AsyncIterable<Uint8Array>>}
    */
   async * fetchContent (cidPath, opts = {}) {
-    const { res, controller, log } = await this.fetchCID(cidPath, opts)
+    let res, controller, log
+
+    if (opts.raceNodes) {
+      ({ res, controller, log } = await this.fetchCIDWithRace(cidPath, opts))
+    } else {
+      ({ res, controller, log } = await this.fetchCID(cidPath, opts))
+    }
 
     async function * metricsIterable (itr) {
       log.numBytesSent = 0
@@ -224,6 +334,7 @@ export class Saturn {
    * @param {string} cidPath
    * @param {object} [opts={}]
    * @param {('car'|'raw')} [opts.format]
+   * @param {boolean} [opts.raceNodes]
    * @param {number} [opts.connectTimeout=5000]
    * @param {number} [opts.downloadTimeout=0]
    * @returns {Promise<Uint8Array>}
@@ -239,7 +350,7 @@ export class Saturn {
    * @returns {URL}
    */
   createRequestURL (cidPath, opts) {
-    let origin = opts.url || opts.cdnURL
+    let origin = opts.url || (opts.origins && opts.origins[0]) || opts.cdnURL
     origin = addHttpPrefix(origin)
     const url = new URL(`${origin}/ipfs/${cidPath}`)
 

test/fallback.spec.js

@@ -128,7 +128,74 @@ describe('Client Fallback', () => {
 
     const saturn = new Saturn({ storage: mockStorage, clientKey: CLIENT_KEY, clientId: 'test' })
 
-    const cid = saturn.fetchContentWithFallback('bafkreifjjcie6lypi6ny7amxnfftagclbuxndqonfipmb64f2km2devei4', { url: 'node1.saturn.ms' })
+    const cid = saturn.fetchContentWithFallback('bafkreifjjcie6lypi6ny7amxnfftagclbuxndqonfipmb64f2km2devei4')
+
+    const buffer = await concatChunks(cid)
+    const actualContent = String.fromCharCode(...buffer)
+    const expectedContent = 'hello world\n'
+
+    assert.strictEqual(actualContent, expectedContent)
+    server.close()
+    mock.reset()
+  })
+
+  test('Content Fallback fetches a cid properly with race', async (t) => {
+    const handlers = [
+      mockOrchHandler(5, TEST_DEFAULT_ORCH, 'saturn.ms'),
+      mockJWT(TEST_AUTH),
+      mockSaturnOriginHandler(TEST_ORIGIN_DOMAIN, 0, true),
+      ...mockNodesHandlers(5, TEST_ORIGIN_DOMAIN)
+    ]
+    const server = getMockServer(handlers)
+    server.listen(MSW_SERVER_OPTS)
+
+    const expectedNodes = generateNodes(3, TEST_ORIGIN_DOMAIN)
+
+    // Mocking storage object
+    const mockStorage = {
+      get: async (key) => expectedNodes,
+      set: async (key, value) => { return null }
+    }
+    t.mock.method(mockStorage, 'get')
+    t.mock.method(mockStorage, 'set')
+
+    const saturn = new Saturn({ storage: mockStorage, clientKey: CLIENT_KEY, clientId: 'test' })
+    // const origins =
+
+    const cid = saturn.fetchContentWithFallback('bafkreifjjcie6lypi6ny7amxnfftagclbuxndqonfipmb64f2km2devei4', { raceNodes: true })
+
+    const buffer = await concatChunks(cid)
+    const actualContent = String.fromCharCode(...buffer)
+    const expectedContent = 'hello world\n'
+
+    assert.strictEqual(actualContent, expectedContent)
+    server.close()
+    mock.reset()
+  })
+
+  test('Content Fallback with race fetches from consecutive nodes on failure', async (t) => {
+    const handlers = [
+      mockOrchHandler(5, TEST_DEFAULT_ORCH, 'saturn.ms'),
+      mockJWT(TEST_AUTH),
+      mockSaturnOriginHandler(TEST_ORIGIN_DOMAIN, 0, true),
+      ...mockNodesHandlers(5, TEST_ORIGIN_DOMAIN, 2)
+    ]
+    const server = getMockServer(handlers)
+    server.listen(MSW_SERVER_OPTS)
+
+    const expectedNodes = generateNodes(5, TEST_ORIGIN_DOMAIN)
+
+    // Mocking storage object
+    const mockStorage = {
+      get: async (key) => expectedNodes,
+      set: async (key, value) => { return null }
+    }
+    t.mock.method(mockStorage, 'get')
+    t.mock.method(mockStorage, 'set')
+
+    const saturn = new Saturn({ storage: mockStorage, clientKey: CLIENT_KEY, clientId: 'test' })
+
+    const cid = saturn.fetchContentWithFallback('bafkreifjjcie6lypi6ny7amxnfftagclbuxndqonfipmb64f2km2devei4', { raceNodes: true })
 
     const buffer = await concatChunks(cid)
     const actualContent = String.fromCharCode(...buffer)

test/test-utils.js

@@ -10,6 +10,7 @@ import fs from 'fs'
 import { addHttpPrefix } from '../src/utils/url.js'
 
 const HTTP_STATUS_OK = 200
+const HTTP_STATUS_TIMEOUT = 504
 
 const __dirname = dirname(fileURLToPath(import.meta.url))
 process.env.TESTING = 'true'
@@ -123,14 +124,23 @@ export function mockJWT (authURL) {
  *
  * @param {number} count - amount of nodes to mock
  * @param {string} originDomain - saturn origin domain.
+ * @param {number} failures
  * @returns {RestHandler<any>[]}
  */
-export function mockNodesHandlers (count, originDomain) {
+export function mockNodesHandlers (count, originDomain, failures = 0) {
+  if (failures > count) {
+    throw Error('failures number cannot exceed node count')
+  }
   const nodes = generateNodes(count, originDomain)
 
-  const handlers = nodes.map((node) => {
+  const handlers = nodes.map((node, idx) => {
     const url = `${node.url}/ipfs/:cid`
     return rest.get(url, (req, res, ctx) => {
+      if (idx < failures) {
+        return res(
+          ctx.status(HTTP_STATUS_TIMEOUT)
+        )
+      }
       const filepath = getFixturePath('hello.car')
       const fileContents = fs.readFileSync(filepath)
       return res(