feat: implement client fallback mechanism

package.json

@@ -6,7 +6,7 @@
   "main": "src/index.js",
   "type": "module",
   "scripts": {
-    "test": "NODE_ENV=development node  --test test/",
+    "test": "NODE_ENV=development TESTING=true node  --test test/",
     "build": "webpack --mode production",
     "prepack": "npm run build"
   },
@@ -30,6 +30,7 @@
     "browser-readablestream-to-it": "^2.0.4",
     "idb": "^7.1.1",
     "ipfs-unixfs-exporter": "https://gitpkg.now.sh/filecoin-saturn/js-ipfs-unixfs/packages/ipfs-unixfs-exporter?build",
+    "msw": "^1.3.2",
     "multiformats": "^12.1.1"
   },
   "devDependencies": {
@@ -41,10 +42,17 @@
     "webpack-cli": "^4.10.0"
   },
   "eslintConfig": {
+    "plugins": [
+      "jsdoc",
+      "promise"
+    ],
     "extends": "ipfs",
     "parserOptions": {
-      "ecmaVersion": "latest",
-      "sourceType": "module"
+      "sourceType": "module",
+      "ecmaVersion": "latest"
+    },
+    "rules": {
+      "jsdoc/no-undefined-types": 1
     }
   },
   "imports": {
@@ -53,4 +61,4 @@
   "publishConfig": {
     "access": "public"
   }
-}
+}

src/client.js

@@ -1,13 +1,17 @@
+// @ts-check
+
 import { CID } from 'multiformats'
 
 import { extractVerifiedContent } from './utils/car.js'
 import { asAsyncIterable, asyncIteratorToBuffer } from './utils/itr.js'
 import { randomUUID } from './utils/uuid.js'
 import { memoryStorage } from './storage/index.js'
 import { getJWT } from './utils/jwt.js'
+import { parseUrl, addHttpPrefix } from './utils/url.js'
 import { isBrowserContext } from './utils/runtime.js'
 
 export class Saturn {
+  static nodesListKey = 'saturn-nodes'
   /**
    *
    * @param {object} [opts={}]
@@ -16,14 +20,18 @@ export class Saturn {
    * @param {string} [opts.cdnURL=saturn.ms]
    * @param {number} [opts.connectTimeout=5000]
    * @param {number} [opts.downloadTimeout=0]
+   * @param {string} [opts.orchURL]
+   * @param {number} [opts.fallbackLimit]
    * @param {import('./storage/index.js').Storage} [opts.storage]
    */
   constructor (opts = {}) {
     this.opts = Object.assign({}, {
       clientId: randomUUID(),
       cdnURL: 'l1s.saturn.ms',
       logURL: 'https://twb3qukm2i654i3tnvx36char40aymqq.lambda-url.us-west-2.on.aws/',
+      orchURL: 'https://orchestrator.strn.pl/nodes?maxNodes=100',
       authURL: 'https://fz3dyeyxmebszwhuiky7vggmsu0rlkoy.lambda-url.us-west-2.on.aws/',
+      fallbackLimit: 5,
       connectTimeout: 5_000,
       downloadTimeout: 0
     }, opts)
@@ -33,13 +41,14 @@ export class Saturn {
     }
 
     this.logs = []
-    this.storage = this.opts.storage || memoryStorage()
+    this.nodes = []
     this.reportingLogs = process?.env?.NODE_ENV !== 'development'
     this.hasPerformanceAPI = isBrowserContext && self?.performance
-
     if (this.reportingLogs && this.hasPerformanceAPI) {
       this._monitorPerformanceBuffer()
     }
+    this.storage = this.opts.storage || memoryStorage()
+    this.loadNodesPromise = this._loadNodes(this.opts)
   }
 
   /**
@@ -76,10 +85,9 @@ export class Saturn {
         Authorization: 'Bearer ' + options.jwt
       }
     }
-
     let res
     try {
-      res = await fetch(url, { signal: controller.signal, ...options })
+      res = await fetch(parseUrl(url), { signal: controller.signal, ...options })
 
       clearTimeout(connectTimeout)
 
@@ -109,6 +117,74 @@ export class Saturn {
     return { res, controller, log }
   }
 
+  /**
+   *
+   * @param {string} cidPath
+   * @param {object} [opts={}]
+   * @param {('car'|'raw')} [opts.format]
+   * @param {string} [opts.url]
+   * @param {number} [opts.connectTimeout=5000]
+   * @param {number} [opts.downloadTimeout=0]
+   * @returns {Promise<AsyncIterable<Uint8Array>>}
+   */
+  async * fetchContentWithFallback (cidPath, opts = {}) {
+    let lastError = null
+    // we use this to checkpoint at which chunk a request failed.
+    // this is temporary until range requests are supported.
+    let byteCountCheckpoint = 0
+
+    const fetchContent = async function * () {
+      let byteCount = 0
+      const byteChunks = await this.fetchContent(cidPath, opts)
+      for await (const chunk of byteChunks) {
+        // avoid sending duplicate chunks
+        if (byteCount < byteCountCheckpoint) {
+          // checks for overlapping chunks
+          const remainingBytes = byteCountCheckpoint - byteCount
+          if (remainingBytes < chunk.length) {
+            yield chunk.slice(remainingBytes)
+            byteCountCheckpoint += chunk.length - remainingBytes
+          }
+        } else {
+          yield chunk
+          byteCountCheckpoint += chunk.length
+        }
+        byteCount += chunk.length
+      }
+    }.bind(this)
+
+    if (this.nodes.length === 0) {
+      // fetch from origin in the case that no nodes are loaded
+      opts.url = this.opts.cdnURL
+      try {
+        yield * fetchContent()
+        return
+      } catch (err) {
+        lastError = err
+        await this.loadNodesPromise
+      }
+    }
+
+    let fallbackCount = 0
+    for (const origin of this.nodes) {
+      if (fallbackCount > this.opts.fallbackLimit) {
+        return
+      }
+      opts.url = origin.url
+      try {
+        yield * fetchContent()
+        return
+      } catch (err) {
+        lastError = err
+      }
+      fallbackCount += 1
+    }
+
+    if (lastError) {
+      throw new Error(`All attempts to fetch content have failed. Last error: ${lastError.message}`)
+    }
+  }
+
   /**
    *
    * @param {string} cidPath
@@ -163,10 +239,8 @@ export class Saturn {
    * @returns {URL}
    */
   createRequestURL (cidPath, opts) {
-    let origin = opts.cdnURL
-    if (!origin.startsWith('http')) {
-      origin = `https://${origin}`
-    }
+    let origin = opts.url || opts.cdnURL
+    origin = addHttpPrefix(origin)
     const url = new URL(`${origin}/ipfs/${cidPath}`)
 
     url.searchParams.set('format', opts.format)
@@ -196,7 +270,6 @@ export class Saturn {
    */
   reportLogs (log) {
     if (!this.reportingLogs) return
-
     this.logs.push(log)
     this.reportLogsTimeout && clearTimeout(this.reportLogsTimeout)
     this.reportLogsTimeout = setTimeout(this._reportLogs.bind(this), 3_000)
@@ -295,4 +368,46 @@ export class Saturn {
       performance.clearResourceTimings()
     }
   }
+
+  async _loadNodes (opts) {
+    let origin = opts.orchURL
+
+    let cacheNodesListPromise
+    if (this.storage) {
+      cacheNodesListPromise = this.storage.get(Saturn.nodesListKey)
+    }
+
+    origin = addHttpPrefix(origin)
+
+    const url = new URL(origin)
+    const controller = new AbortController()
+    const options = Object.assign({}, { method: 'GET' }, this.opts)
+
+    const connectTimeout = setTimeout(() => {
+      controller.abort()
+    }, options.connectTimeout)
+
+    const orchResponse = await fetch(parseUrl(url), { signal: controller.signal, ...options })
+    const orchNodesListPromise = orchResponse.json()
+    clearTimeout(connectTimeout)
+
+    // This promise races fetching nodes list from the orchestrator and
+    // and the provided storage object (localStorage, sessionStorage, etc.)
+    // to insure we have a fallback set as quick as possible
+    let nodes
+    if (cacheNodesListPromise) {
+      nodes = await Promise.any([orchNodesListPromise, cacheNodesListPromise])
+    } else {
+      nodes = await orchNodesListPromise
+    }
+
+    // if storage returns first, update based on cached storage.
+    if (nodes === await cacheNodesListPromise) {
+      this.nodes = nodes
+    }
+    // we always want to update from the orchestrator regardless.
+    nodes = await orchNodesListPromise
+    this.nodes = nodes
+    this.storage?.set(Saturn.nodesListKey, nodes)
+  }
 }

src/utils/url.js

@@ -0,0 +1,32 @@
+// @ts-check
+
+/**
+ *
+ * @param {URL} url
+ * @returns {URL|string}
+ */
+export function parseUrl (url) {
+  try {
+    // This is a temp function to resolve URLs for mock testing
+  // See issue here: https://github.com/mswjs/msw/issues/1597
+    if (process?.env?.TESTING) {
+      return url.toJSON()
+    }
+  } catch (e) {}
+
+  return url
+}
+
+/**
+ *
+ * @param {string} url
+ * @returns {string}
+ */
+export function addHttpPrefix (url) {
+  // This is a temp function to resolve URLs for mock testing
+  // See issue here: https://github.com/mswjs/msw/issues/1597
+  if (!url.startsWith('http')) {
+    url = `https://${url}`
+  }
+  return url
+}

test/car.spec.js

@@ -1,28 +1,13 @@
 import assert from 'node:assert/strict'
 import fs from 'node:fs'
 import { describe, it } from 'node:test'
-import { resolve, dirname } from 'node:path'
-import { fileURLToPath } from 'node:url'
+import { getFixturePath, concatChunks } from './test-utils.js'
 
 import { CarReader, CarWriter } from '@ipld/car'
 import { CID } from 'multiformats/cid'
 
 import { extractVerifiedContent } from '#src/utils/car.js'
 
-const __dirname = dirname(fileURLToPath(import.meta.url))
-
-function getFixturePath (filename) {
-  return resolve(__dirname, `./fixtures/${filename}`)
-}
-
-async function concatChunks (itr) {
-  const arr = []
-  for await (const chunk of itr) {
-    arr.push(chunk)
-  }
-  return new Uint8Array(...arr)
-}
-
 describe('CAR Verification', () => {
   it('should extract content from a valid CAR', async () => {
     const cidPath =
@@ -149,8 +134,7 @@ describe('CAR Verification', () => {
 
     await assert.rejects(
       async () => {
-        for await (const _ of extractVerifiedContent(cidPath, out)) {
-        }
+        for await (const _ of extractVerifiedContent(cidPath, out)) {}
       },
       {
         name: 'VerificationError',