Add alternative provider retrieval check

lib/ipni-client.js

@@ -5,12 +5,22 @@ import {
   assertOkResponse,
 } from '../vendor/deno-deps.js'
 
+/**
+ * @typedef {{
+ *   peerId: string
+ *   address: string
+ *   protocol: string
+ *   contextId: string
+ * }} Provider
+ */
+
 /**
  * @param {string} cid
  * @param {string} providerId
  * @returns {Promise<{
  *   indexerResult: string
- *   provider?: { address: string; protocol: string }
+ *   provider?: Provider
+ *   alternativeProviders?: Provider[]
  * }>}
  */
 export async function queryTheIndex(cid, providerId) {
@@ -38,9 +48,8 @@ export async function queryTheIndex(cid, providerId) {
   }
 
   let graphsyncProvider
+  const alternativeProviders = []
   for (const p of providerResults) {
-    if (p.Provider.ID !== providerId) continue
-
     const [protocolCode] = decodeVarint(decodeBase64(p.Metadata))
     const protocol = {
       0x900: 'bitswap',
@@ -52,22 +61,32 @@ export async function queryTheIndex(cid, providerId) {
     const address = p.Provider.Addrs[0]
     if (!address) continue
 
+    const provider = {
+      peerId: p.Provider.ID,
+      address: formatProviderAddress(p.Provider.ID, address, protocol),
+      contextId: p.ContextID,
+      protocol,
+    }
+
+    if (p.Provider.ID !== providerId) {
+      alternativeProviders.push(provider)
+      continue
+    }
+
     switch (protocol) {
       case 'http':
         return {
           indexerResult: 'OK',
-          provider: { address, protocol },
+          provider,
         }
 
       case 'graphsync':
         if (!graphsyncProvider) {
-          graphsyncProvider = {
-            address: `${address}/p2p/${p.Provider.ID}`,
-            protocol,
-          }
+          graphsyncProvider = provider
         }
     }
   }
+
   if (graphsyncProvider) {
     console.log('HTTP protocol is not advertised, falling back to Graphsync.')
     return {
@@ -79,7 +98,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', alternativeProviders }
 }
 
 async function getRetrievalProviders(cid) {
@@ -90,3 +109,7 @@ async function getRetrievalProviders(cid) {
   const result = await res.json()
   return result.MultihashResults.flatMap((r) => r.ProviderResults)
 }
+
+function formatProviderAddress(peerId, address, protocol) {
+  return protocol === 'http' ? address : `${address}/p2p/${peerId}`
+}

lib/spark.js

@@ -1,5 +1,6 @@
 /* global Zinnia */
 
+/** @import {Provider} from './ipni-client.js' */
 import { ActivityState } from './activity-state.js'
 import {
   SPARK_VERSION,
@@ -44,18 +45,20 @@ export default class Spark {
 
   async getRetrieval() {
     const retrieval = await this.#tasker.next()
-    if (retrieval) {
-      console.log({ retrieval })
+    const { retrievalTask } = retrieval
+    if (retrievalTask) {
+      console.log({ retrievalTask })
     }
+
     return retrieval
   }
 
-  async executeRetrievalCheck(retrieval, stats) {
+  async executeRetrievalCheck({ retrievalTask, stats, randomness }) {
     console.log(
-      `Calling Filecoin JSON-RPC to get PeerId of miner ${retrieval.minerId}`,
+      `Calling Filecoin JSON-RPC to get PeerId of miner ${retrievalTask.minerId}`,
     )
     try {
-      const peerId = await this.#getIndexProviderPeerId(retrieval.minerId)
+      const peerId = await this.#getIndexProviderPeerId(retrievalTask.minerId)
       console.log(`Found peer id: ${peerId}`)
       stats.providerId = peerId
     } catch (err) {
@@ -78,29 +81,45 @@ export default class Spark {
     }
 
     console.log(
-      `Querying IPNI to find retrieval providers for ${retrieval.cid}`,
-    )
-    const { indexerResult, provider } = await queryTheIndex(
-      retrieval.cid,
-      stats.providerId,
+      `Querying IPNI to find retrieval providers for ${retrievalTask.cid}`,
     )
+    const { indexerResult, provider, alternativeProviders } =
+      await queryTheIndex(retrievalTask.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 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...',
+      )
+      stats.alternativeProviderCheck =
+        await this.checkRetrievalFromAlternativeProvider({
+          alternativeProviders,
+          randomness,
+          cid: retrievalTask.cid,
+        })
+      return
+    }
 
     stats.protocol = provider.protocol
     stats.providerAddress = provider.address
 
     await this.fetchCAR(
       provider.protocol,
       provider.address,
-      retrieval.cid,
+      retrievalTask.cid,
       stats,
     )
     if (stats.protocol === 'http') {
-      await this.testHeadRequest(provider.address, retrieval.cid, stats)
+      await this.testHeadRequest(provider.address, retrievalTask.cid, stats)
     }
   }
 
@@ -220,6 +239,51 @@ export default class Spark {
     }
   }
 
+  async checkRetrievalFromAlternativeProvider({
+    alternativeProviders,
+    randomness,
+    cid,
+  }) {
+    if (!alternativeProviders.length) {
+      console.info('No alternative providers found for this CID.')
+      return
+    }
+
+    const validAlternativeProviders = alternativeProviders.filter(
+      (p) => p.protocol !== 'bitswap',
+    )
+
+    if (!validAlternativeProviders.length) {
+      console.warn(
+        'No providers serving the content via HTTP or Graphsync found. Skipping network-wide retrieval check.',
+      )
+      return
+    }
+
+    const alternativeProvider = pickRandomProvider(
+      validAlternativeProviders,
+      randomness,
+    )
+    if (!alternativeProvider) {
+      console.warn(
+        'Failed to pick a alternative provider. Skipping network-wide retrieval check.',
+      )
+      return
+    }
+
+    const alternativeProviderRetrievalStats = newAlternativeProviderCheckStats()
+    alternativeProviderRetrievalStats.providerId = alternativeProvider.peerId
+
+    await this.fetchCAR(
+      alternativeProvider.protocol,
+      alternativeProvider.address,
+      cid,
+      alternativeProviderRetrievalStats,
+    )
+
+    return alternativeProviderRetrievalStats
+  }
+
   async submitMeasurement(task, stats) {
     console.log('Submitting measurement...')
     const payload = {
@@ -246,8 +310,8 @@ export default class Spark {
   }
 
   async nextRetrieval() {
-    const retrieval = await this.getRetrieval()
-    if (!retrieval) {
+    const { retrievalTask, randomness } = await this.getRetrieval()
+    if (!retrievalTask) {
       console.log(
         'Completed all tasks for the current round. Waiting for the next round to start.',
       )
@@ -256,9 +320,11 @@ export default class Spark {
 
     const stats = newStats()
 
-    await this.executeRetrievalCheck(retrieval, stats)
+    await this.executeRetrievalCheck({ retrievalTask, randomness, stats })
 
-    const measurementId = await this.submitMeasurement(retrieval, { ...stats })
+    const measurementId = await this.submitMeasurement(retrievalTask, {
+      ...stats,
+    })
     Zinnia.jobCompleted()
     return measurementId
   }
@@ -335,6 +401,17 @@ export function newStats() {
     carChecksum: null,
     statusCode: null,
     headStatusCode: null,
+    alternativeProviderCheck: null,
+  }
+}
+
+function newAlternativeProviderCheckStats() {
+  return {
+    statusCode: null,
+    timeout: false,
+    endAt: null,
+    carTooLarge: false,
+    providerId: null,
   }
 }
 
@@ -418,3 +495,64 @@ function mapErrorToStatusCode(err) {
   // Fallback code for unknown errors
   return 600
 }
+
+/**
+ * Picks a random provider based on their priority and supplied randomness.
+ *
+ * Providers are prioritized in the following order:
+ *
+ * 1. HTTP Providers with Piece Info advertised in their ContextID.
+ * 2. Graphsync Providers with Piece Info advertised in their ContextID.
+ * 3. HTTP Providers.
+ * 4. Graphsync Providers.
+ *
+ * @param {Provider[]} providers
+ * @param {string} randomness
+ * @returns {Provider | undefined}
+ */
+export function pickRandomProvider(providers, randomness) {
+  const filterByProtocol = (items, protocol) =>
+    items.filter((provider) => provider.protocol === protocol)
+
+  const pickRandomItem = (items) => {
+    if (!items.length) return undefined
+    const randomValue = BigInt('0x' + randomness)
+    const ix = Number(randomValue % BigInt(items.length))
+    return items[ix]
+  }
+
+  const providersWithPieceInfoContextID = providers.filter((p) =>
+    p.contextId.startsWith('ghsA'),
+  )
+
+  // Priority 1: HTTP providers with ContextID containing PieceCID
+  const httpProvidersWithPieceInfoContextID = filterByProtocol(
+    providersWithPieceInfoContextID,
+    'http',
+  )
+  if (httpProvidersWithPieceInfoContextID.length) {
+    return pickRandomItem(httpProvidersWithPieceInfoContextID, randomness)
+  }
+
+  // Priority 2: Graphsync providers with ContextID containing PieceCID
+  const graphsyncProvidersWithPieceInfoContextID = filterByProtocol(
+    providersWithPieceInfoContextID,
+    'graphsync',
+  )
+  if (graphsyncProvidersWithPieceInfoContextID.length) {
+    return pickRandomItem(graphsyncProvidersWithPieceInfoContextID, randomness)
+  }
+
+  // Priority 3: HTTP providers
+  const httpProviders = filterByProtocol(providers, 'http')
+  if (httpProviders.length) return pickRandomItem(httpProviders, randomness)
+
+  // Priority 4: Graphsync providers
+  const graphsyncProviders = filterByProtocol(providers, 'graphsync')
+  if (graphsyncProviders.length) {
+    return pickRandomItem(graphsyncProviders, randomness)
+  }
+
+  // No valid providers found
+  return undefined
+}

lib/tasker.js

@@ -15,6 +15,7 @@ export class Tasker {
   #remainingRoundTasks
   #fetch
   #activity
+  #randomness
 
   /**
    * @param {object} args
@@ -37,10 +38,11 @@ export class Tasker {
     this.#remainingRoundTasks = []
   }
 
-  /** @returns {Task | undefined} */
+  /** @returns {Promise<{ retrievalTask?: RetrievalTask; randomness: string }>} */
   async next() {
     await this.#updateCurrentRound()
-    return this.#remainingRoundTasks.pop()
+    const retrievalTask = this.#remainingRoundTasks.pop()
+    return { retrievalTask, randomness: this.#randomness }
   }
 
   async #updateCurrentRound() {
@@ -76,13 +78,13 @@ export class Tasker {
     console.log('  %s retrieval tasks', retrievalTasks.length)
     this.maxTasksPerRound = maxTasksPerNode
 
-    const randomness = await getRandomnessForSparkRound(round.startEpoch)
-    console.log('  randomness: %s', randomness)
+    this.#randomness = await getRandomnessForSparkRound(round.startEpoch)
+    console.log('  randomness: %s', this.#randomness)
 
     this.#remainingRoundTasks = await pickTasksForNode({
       tasks: retrievalTasks,
       maxTasksPerRound: this.maxTasksPerRound,
-      randomness,
+      randomness: this.#randomness,
       stationId: Zinnia.stationId,
     })