Add alternative provider retrieval check

lib/ipni-client.js

@@ -1,12 +1,15 @@
 import { decodeBase64, decodeVarint, pRetry, assertOkResponse } from '../vendor/deno-deps.js'
 
+/** @typedef {{ address: string; protocol: string; contextId: string; }} Provider  */
+
 /**
  *
  * @param {string} cid
  * @param {string} providerId
  * @returns {Promise<{
  *  indexerResult: string;
- *  provider?: { address: string; protocol: string };
+ *  provider?: Provider;
+ *  providers?: Provider[];
  * }>}
  */
 export async function queryTheIndex(cid, providerId) {
@@ -31,9 +34,8 @@ export async function queryTheIndex(cid, providerId) {
   }
 
   let graphsyncProvider
+  const providers = []
   for (const p of providerResults) {
-    if (p.Provider.ID !== providerId) continue
-
     const [protocolCode] = decodeVarint(decodeBase64(p.Metadata))
     const protocol = {
       0x900: 'bitswap',
@@ -45,22 +47,33 @@ export async function queryTheIndex(cid, providerId) {
     const address = p.Provider.Addrs[0]
     if (!address) continue
 
-    switch (protocol) {
-      case 'http':
-        return {
-          indexerResult: 'OK',
-          provider: { address, protocol },
-        }
+    const provider = {
+      address: formatProviderAddress(p.Provider.ID, address, protocol),
+      contextId: p.ContextID,
+      protocol,
+    }
+
+    if (p.Provider.ID === providerId) {
+      switch (protocol) {
+        case 'http':
+          return {
+            indexerResult: 'OK',
+            provider,
+          }
 
-      case 'graphsync':
-        if (!graphsyncProvider) {
-          graphsyncProvider = {
-            address: `${address}/p2p/${p.Provider.ID}`,
-            protocol,
+        case 'graphsync':
+          if (!graphsyncProvider) {
+            graphsyncProvider = provider
           }
-        }
+      }
+
+      // Skip adding the provider to the list if it's the the one we are looking for
+      continue
     }
+
+    providers.push(provider)
   }
+
   if (graphsyncProvider) {
     console.log('HTTP protocol is not advertised, falling back to Graphsync.')
     return {
@@ -70,7 +83,7 @@ export async function queryTheIndex(cid, providerId) {
   }
 
   console.log('All advertisements are from other miners or for unsupported protocols.')
-  return { indexerResult: 'NO_VALID_ADVERTISEMENT' }
+  return { indexerResult: 'NO_VALID_ADVERTISEMENT', providers }
 }
 
 async function getRetrievalProviders(cid) {
@@ -81,3 +94,7 @@ async function getRetrievalProviders(cid) {
   const result = await res.json()
   return result.MultihashResults.flatMap((r) => r.ProviderResults)
 }
+
+function formatProviderAddress(id, address, protocol) {
+  return protocol === 'http' ? address : `${address}/p2p/${id}`
+}

lib/spark.js

@@ -1,5 +1,6 @@
 /* global Zinnia */
 
+/** @import { Provider } from './ipni-client.js' */
 import { ActivityState } from './activity-state.js'
 import {
   SPARK_VERSION,
@@ -71,11 +72,25 @@ export default class Spark {
     }
 
     console.log(`Querying IPNI to find retrieval providers for ${retrieval.cid}`)
-    const { indexerResult, provider } = await queryTheIndex(retrieval.cid, stats.providerId)
+    const { indexerResult, provider, providers } = await queryTheIndex(
+      retrieval.cid,
+      stats.providerId,
+    )
     stats.indexerResult = indexerResult
 
     const providerFound = indexerResult === 'OK' || indexerResult === 'HTTP_NOT_ADVERTISED'
-    if (!providerFound) return
+    const noValidAdvertisement = indexerResult === 'NO_VALID_ADVERTISEMENT'
+
+    // In case index lookup failed due to network error or CID not found,
+    // we will not perform any retrieval
+    if (!providerFound && !noValidAdvertisement) return
+
+    // In case we fail to find a valid advertisement for the provider
+    // we will try to perform network wide retrieval from other providers
+    if (noValidAdvertisement) {
+      console.log('No valid advertisement found. Trying to retrieve from an alternative provider...')
+      return await this.checkRetrievalFromAlternativeProvider(providers, retrieval.cid, stats)
+    }
 
     stats.protocol = provider.protocol
     stats.providerAddress = provider.address
@@ -202,6 +217,29 @@ export default class Spark {
     }
   }
 
+  async testNetworkRetrieval(providers, cid, stats) {
+    if (!providers.length) {
+      console.info('No alternative providers found for the CID.')
+      return
+    }
+
+    stats.networkRetrieval = newNetworkRetrievalStats()
+    const randomProvider = pickRandomProvider(providers)
+    if (!randomProvider) {
+      console.warn(
+        'No providers serving the content via HTTP or Graphsync found. Skipping network-wide retrieval check.',
+      )
+      return
+    }
+
+    await this.fetchCAR(
+      randomProvider.protocol,
+      randomProvider.address,
+      cid,
+      stats.networkRetrieval,
+    )
+  }
+
   async submitMeasurement(task, stats) {
     console.log('Submitting measurement...')
     const payload = {
@@ -315,6 +353,16 @@ export function newStats() {
     carChecksum: null,
     statusCode: null,
     headStatusCode: null,
+    networkRetrieval: null,
+  }
+}
+
+function newNetworkRetrievalStats() {
+  return {
+    statusCode: null,
+    timeout: false,
+    endAt: null,
+    carTooLarge: false,
   }
 }
 
@@ -395,3 +443,65 @@ function mapErrorToStatusCode(err) {
   // Fallback code for unknown errors
   return 600
 }
+
+/**
+ * Assigns weights to providers based on their protocol and context ID and picks one at random.
+ * Providers with higher weights have a higher chance of being selected.
+ *
+ * Providers serving content using Bitswap protocol are filtered out.
+ *
+ * @param {Provider[]} providers
+ * @returns {Provider | undefined}
+ */
+export function pickRandomProvider(providers) {
+  const filteredProviders = providers.filter((provider) => provider.protocol !== 'bitswap')
+  if (!filteredProviders.length) return
+  const weightedProviders = weighProviders(filteredProviders)
+  return pickRandomWeightedItem(weightedProviders)
+}
+
+/**
+ * Assigns weights to providers based on their protocol and context ID.
+ *
+ * HTTP providers and those whose context ID starts with 'gHa' are given higher weights,
+ * hence having a higher chance of being selected.
+ *
+ * @param {Provider[]} providers
+ * @returns {Provider & { weight: number }[]}
+ */
+function weighProviders(providers) {
+  const protocolWeights = { http: 2, graphsync: 1 }
+
+  // assign weight to each provider
+  return providers.map((provider) => {
+    let weight = protocolWeights[provider.protocol]
+    if (provider.contextId.startsWith('ghsA')) weight += 1
+
+    return {
+      ...provider,
+      weight,
+    }
+  })
+}
+
+/**
+ * Picks a random item from an array based on their weight. The higher the weight, the higher the chance of being selected.
+ *
+ * @template T The type of the item in the list.
+ * @param {Array<{weight: number}>} items The list of items, where each item has a `weight`property.
+ * @returns {T} The randomly selected item based on its weight.
+ *
+ */
+function pickRandomWeightedItem(items) {
+  const totalWeight = items.reduce((acc, item) => acc + item.weight, 0)
+  let random = Math.random() * totalWeight
+
+  // Iterate over items, subtracting the item's weight from the random number
+  // until we find the item where the random number is less than the item's weight
+  for (let i = 0; i < items.length; i++) {
+    random -= items[i].weight
+    if (random <= 0) {
+      return items[i]
+    }
+  }
+}

test/ipni-client.test.js

@@ -12,6 +12,7 @@ test('query advertised CID', async () => {
     provider: {
       address: '/dns/frisbii.fly.dev/tcp/443/https',
       protocol: 'http',
+      contextId: 'ZnJpc2JpaQ==',
     },
   })
 })

test/spark.js

@@ -1,6 +1,6 @@
 /* global Zinnia */
 
-import Spark, { calculateDelayBeforeNextTask, newStats } from '../lib/spark.js'
+import Spark, { calculateDelayBeforeNextTask, newStats, pickRandomProvider } from '../lib/spark.js'
 import { test } from 'zinnia:test'
 import {
   assertInstanceOf,
@@ -9,6 +9,7 @@ import {
   assertNotEquals,
   assertLessOrEqual,
   assertGreaterOrEqual,
+  assertGreater,
 } from 'zinnia:assert'
 import { SPARK_VERSION } from '../lib/constants.js'
 
@@ -155,6 +156,60 @@ test('fetchCAR - http', async () => {
     'stats.carChecksum',
   )
   assertEquals(requests, [`https://frisbii.fly.dev/ipfs/${KNOWN_CID}?dag-scope=block`])
+  assertEquals(stats.networkRetrieval, null, 'stats.networkRetrieval')
+})
+
+test('testNetworkRetrieval - http', async () => {
+  const requests = []
+  const spark = new Spark({
+    fetch: async (url) => {
+      requests.push(url.toString())
+      return fetch(url)
+    },
+  })
+  const stats = newStats()
+  const providers = [
+    {
+      address: '/dns/frisbii.fly.dev/tcp/443/https',
+      protocol: 'http',
+      contextId: 'ZnJpc2JpaQ==',
+    },
+    {
+      address: '/dns/mock.fly.dev/tcp/443/https',
+      protocol: 'bitswap',
+      contextId: 'ghA==',
+    },
+  ]
+
+  await spark.testNetworkRetrieval(providers, KNOWN_CID, stats)
+  assertEquals(stats.networkRetrieval.statusCode, 200, 'stats.networkRetrieval.statusCode')
+  assertEquals(stats.networkRetrieval.timeout, false, 'stats.networkRetrieval.timeout')
+  assertInstanceOf(stats.networkRetrieval.endAt, Date, 'stats.networkRetrieval.endAt')
+  assertEquals(stats.networkRetrieval.carTooLarge, false, 'stats.networkRetrieval.carTooLarge')
+  assertEquals(stats.byteLength, 0, 'stats.byteLength')
+  assertEquals(stats.carChecksum, null, 'stats.carChecksum')
+  assertEquals(requests, [`https://frisbii.fly.dev/ipfs/${KNOWN_CID}?dag-scope=block`])
+  assertEquals(stats.statusCode, null, 'stats.statusCode')
+  assertEquals(stats.timeout, false, 'stats.timeout')
+  assertEquals(stats.startAt, null, 'stats.startAt')
+  assertEquals(stats.firstByteAt, null, 'stats.firstByteAt')
+  assertEquals(stats.endAt, null, 'stats.endAt')
+  assertEquals(stats.carTooLarge, false, 'stats.carTooLarge')
+})
+
+test('testNetworkRetrieval - no providers', async () => {
+  const requests = []
+  const spark = new Spark({
+    fetch: async (url) => {
+      requests.push(url.toString())
+      return fetch(url)
+    },
+  })
+  const stats = newStats()
+  const providers = []
+
+  await spark.testNetworkRetrieval(providers, KNOWN_CID, stats)
+  assertEquals(stats.networkRetrieval, null, 'stats.networkRetrieval')
 })
 
 /* Fixme: Find an active deal on a reliable graphsync provider
@@ -483,3 +538,36 @@ test('fetchCAR triggers timeout after long retrieval', async () => {
 
   assertEquals(stats.timeout, true)
 })
+
+const mockProviders = [
+  { protocol: 'http', contextId: 'ghsA123', address: 'provider1' },
+  { protocol: 'graphsync', contextId: 'ghsB456', address: 'provider2' },
+  { protocol: 'bitswap', contextId: 'ghsC789', address: 'provider3' },
+  // Serves using HTTP but contextId does not start with 'ghsA'
+  { protocol: 'http', contextId: 'ghsB987', address: 'provider4' },
+]
+
+test('pickRandomProvider - should filter out providers using the Bitswap protocol', () => {
+  const result = pickRandomProvider(mockProviders)
+  assertNotEquals(result.protocol, 'bitswap')
+})
+
+test('pickRandomProvider - should return undefined if no providers are left after filtering', () => {
+  const providers = [{ protocol: 'bitswap', contextId: 'ghsC789', address: 'provider3' }]
+  const result = pickRandomProvider(providers)
+  assertEquals(result, undefined)
+})
+
+test('pickRandomProvider - should return a provider with higher weight more frequently', () => {
+  const results = {}
+  for (let i = 0; i < 1000; i++) {
+    const result = pickRandomProvider(mockProviders)
+    if (result) {
+      results[result.address] = (results[result.address] || 0) + 1
+    }
+  }
+
+  // Providers with protocol 'http' and contextId starting with 'ghsA' should have higher counts
+  assertGreater(results['provider1'], results['provider2'])
+  assertGreater(results['provider4'], results['provider2'])
+})