feat: use IPNI advertisements from the miner only

lib/constants.js

@@ -1,3 +1,9 @@
 export const SPARK_VERSION = '1.9.1'
 export const MAX_CAR_SIZE = 200 * 1024 * 1024 // 200 MB
 export const APPROX_ROUND_LENGTH_IN_MS = 20 * 60_000 // 20 minutes
+
+export const RPC_REQUEST = new Request('https://api.node.glif.io/', {
+  headers: {
+    authorization: 'Bearer 6bbc171ebfdd78b2644602ce7463c938'
+  }
+})

lib/ipni-client.js

@@ -3,12 +3,13 @@ import { decodeBase64, decodeVarint } from '../vendor/deno-deps.js'
 /**
  *
  * @param {string} cid
+ * @param {string} providerId
  * @returns {Promise<{
  *  indexerResult: string;
  *  provider?: { address: string; protocol: string };
  * }>}
  */
-export async function queryTheIndex (cid) {
+export async function queryTheIndex (cid, providerId) {
   const url = `https://cid.contact/cid/${encodeURIComponent(cid)}`
 
   let providerResults
@@ -29,11 +30,7 @@ export async function queryTheIndex (cid) {
 
   let graphsyncProvider
   for (const p of providerResults) {
-    // TODO: find only the contact advertised by the SP handling this deal
-    // See https://filecoinproject.slack.com/archives/C048DLT4LAF/p1699958601915269?thread_ts=1699956597.137929&cid=C048DLT4LAF
-    // bytes of CID of dag-cbor encoded DealProposal
-    // https://github.com/filecoin-project/boost/blob/main/indexprovider/wrapper.go#L168-L172
-    // https://github.com/filecoin-project/boost/blob/main/indexprovider/wrapper.go#L195
+    if (p.Provider.ID !== providerId) continue
 
     const [protocolCode] = decodeVarint(decodeBase64(p.Metadata))
     const protocol = {

lib/miner-info.js

@@ -0,0 +1,50 @@
+import { RPC_REQUEST } from './constants.js'
+
+/**
+ * @param {string} minerId A miner actor id, e.g. `f0142637`
+ * @returns {Promise<string>} Miner's PeerId, e.g. `12D3KooWMsPmAA65yHAHgbxgh7CPkEctJHZMeM3rAvoW8CZKxtpG`
+ */
+export async function getMinerPeerId (minerId) {
+  try {
+    const res = await rpc('Filecoin.StateMinerInfo', minerId, null)
+    return res.PeerId
+  } catch (err) {
+    err.message = `Cannot obtain miner info for ${minerId}: ${err.message}`
+    throw err
+  }
+}
+
+/**
+ * @param {string} method
+ * @param {unknown[]} params
+ */
+async function rpc (method, ...params) {
+  const req = new Request(RPC_REQUEST, {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+      accepts: 'application/json'
+    },
+    body: JSON.stringify({
+      jsonrpc: '2.0',
+      id: 1,
+      method,
+      params
+    })
+  })
+  const res = await fetch(req)
+
+  if (!res.ok) {
+    throw new Error(`JSON RPC failed with ${res.code}: ${await res.text()}`)
+  }
+
+  const body = await res.json()
+  if (body.error) {
+    const err = new Error(body.error.message)
+    err.name = 'FilecoinRpcError'
+    err.code = body.code
+    throw err
+  }
+
+  return body.result
+}

lib/spark.js

@@ -3,6 +3,7 @@
 import { ActivityState } from './activity-state.js'
 import { SPARK_VERSION, MAX_CAR_SIZE, APPROX_ROUND_LENGTH_IN_MS } from './constants.js'
 import { queryTheIndex } from './ipni-client.js'
+import { getMinerPeerId as defaultGetMinerPeerId } from './miner-info.js'
 import {
   encodeHex
 } from '../vendor/deno-deps.js'
@@ -11,11 +12,16 @@ const sleep = dt => new Promise(resolve => setTimeout(resolve, dt))
 
 export default class Spark {
   #fetch
+  #getMinerPeerId
   #activity = new ActivityState()
   #maxTasksPerNode = 360
 
-  constructor ({ fetch = globalThis.fetch } = {}) {
+  constructor ({
+    fetch = globalThis.fetch,
+    getMinerPeerId = defaultGetMinerPeerId
+  } = {}) {
     this.#fetch = fetch
+    this.#getMinerPeerId = getMinerPeerId
   }
 
   async getRetrieval () {
@@ -39,8 +45,34 @@ export default class Spark {
   }
 
   async executeRetrievalCheck (retrieval, stats) {
+    console.log(`Calling Filecoin JSON-RPC to get PeerId of miner ${retrieval.minerId}`)
+    try {
+      const peerId = await this.#getMinerPeerId(retrieval.minerId)
+      console.log(`Found peer id: ${peerId}`)
+      stats.providerId = peerId
+    } catch (err) {
+      console.error(err)
+      // There are three common error cases:
+      //  1. We are offline
+      //  2. The JSON RPC provider is down
+      //  3. JSON RPC errors like when Miner ID is not a known actor
+      // There isn't much we can do in the first two cases. We can notify the user that we are not
+      // performing any jobs and wait until the problem is resolved.
+      // The third case should not happen unless we made a mistake, so we want to learn about it
+      if (err.name === 'FilecoinRpcError') {
+        // TODO: report the error to Sentry
+        console.error('The error printed above was not expected, please report it on GitHub:')
+        console.error('https://github.com/filecoin-station/spark/issues/new')
+      } else {
+        this.#activity.onError()
+      }
+      err.reported = true
+      // Abort the check, no measurement should be recorded
+      throw err
+    }
+
     console.log(`Querying IPNI to find retrieval providers for ${retrieval.cid}`)
-    const { indexerResult, provider } = await queryTheIndex(retrieval.cid)
+    const { indexerResult, provider } = await queryTheIndex(retrieval.cid, stats.providerId)
     stats.indexerResult = indexerResult
 
     const providerFound = indexerResult === 'OK' || indexerResult === 'HTTP_NOT_ADVERTISED'
@@ -171,6 +203,7 @@ export default class Spark {
       byteLength: 0,
       carChecksum: null,
       statusCode: null,
+      providerId: null,
       indexerResult: null
     }
 
@@ -188,12 +221,7 @@ export default class Spark {
         await this.nextRetrieval()
         this.#activity.onHealthy()
       } catch (err) {
-        if (err.statusCode === 400 && err.serverMessage === 'OUTDATED CLIENT') {
-          this.#activity.onOutdatedClient()
-        } else {
-          this.#activity.onError()
-        }
-        console.error(err)
+        this.handleRunError(err)
       }
       const duration = Date.now() - started
       const baseDelay = APPROX_ROUND_LENGTH_IN_MS / this.#maxTasksPerNode
@@ -204,6 +232,17 @@ export default class Spark {
       }
     }
   }
+
+  handleRunError (err) {
+    if (err.reported) return
+
+    if (err.statusCode === 400 && err.serverMessage === 'OUTDATED CLIENT') {
+      this.#activity.onOutdatedClient()
+    } else {
+      this.#activity.onError()
+    }
+    console.error(err)
+  }
 }
 
 async function assertOkResponse (res, errorMsg) {

test.js

@@ -1,3 +1,4 @@
 import './test/ipni-client.test.js'
+import './test/miner-info.test.js'
 import './test/integration.js'
 import './test/spark.js'

test/integration.js

@@ -1,8 +1,11 @@
 import Spark from '../lib/spark.js'
-import { test } from 'zinnia:test'
+
 import { assert, assertEquals } from 'zinnia:assert'
+import { test } from 'zinnia:test'
 
 const KNOWN_CID = 'bafkreih25dih6ug3xtj73vswccw423b56ilrwmnos4cbwhrceudopdp5sq'
+const OUR_FAKE_MINER_ID = 'f01spark'
+const FRISBEE_PEER_ID = '12D3KooWN3zbfCjLrjBB7uxYThRTCFM9nxinjb5j9fYFZ6P5RUfP'
 
 test('integration', async () => {
   const spark = new Spark()
@@ -15,15 +18,22 @@ test('integration', async () => {
 })
 
 test('retrieval check for our CID', async () => {
-  const spark = new Spark()
-  spark.getRetrieval = async () => ({ cid: KNOWN_CID })
+  const getMinerPeerId = async (minerId) => {
+    assertEquals(minerId, OUR_FAKE_MINER_ID)
+    return FRISBEE_PEER_ID
+  }
+  const spark = new Spark({ getMinerPeerId })
+  spark.getRetrieval = async () => ({ cid: KNOWN_CID, minerId: OUR_FAKE_MINER_ID })
+
   const measurementId = await spark.nextRetrieval()
   const res = await fetch(`https://api.filspark.com/measurements/${measurementId}`)
   assert(res.ok)
   const m = await res.json()
   const assertProp = (prop, expectedValue) => assertEquals(m[prop], expectedValue, prop)
 
   assertProp('cid', KNOWN_CID)
+  assertProp('minerId', OUR_FAKE_MINER_ID)
+  assertProp('providerId', FRISBEE_PEER_ID)
   assertProp('indexerResult', 'OK')
   assertProp('providerAddress', '/dns/frisbii.fly.dev/tcp/443/https')
   assertProp('protocol', 'http')

test/ipni-client.test.js

@@ -3,9 +3,10 @@ import { assertEquals } from 'zinnia:assert'
 import { queryTheIndex } from '../lib/ipni-client.js'
 
 const KNOWN_CID = 'bafkreih25dih6ug3xtj73vswccw423b56ilrwmnos4cbwhrceudopdp5sq'
+const FRISBEE_PEER_ID = '12D3KooWN3zbfCjLrjBB7uxYThRTCFM9nxinjb5j9fYFZ6P5RUfP'
 
 test('query advertised CID', async () => {
-  const result = await queryTheIndex(KNOWN_CID)
+  const result = await queryTheIndex(KNOWN_CID, FRISBEE_PEER_ID)
   assertEquals(result, {
     indexerResult: 'OK',
     provider: {
@@ -14,3 +15,8 @@ test('query advertised CID', async () => {
     }
   })
 })
+
+test('ignore advertisements from other miners', async () => {
+  const result = await queryTheIndex(KNOWN_CID, '12D3KooWsomebodyelse')
+  assertEquals(result.indexerResult, 'NO_VALID_ADVERTISEMENT')
+})

test/miner-info.test.js

@@ -0,0 +1,21 @@
+import { test } from 'zinnia:test'
+import { assertMatch, AssertionError } from 'zinnia:assert'
+import { getMinerPeerId } from '../lib/miner-info.js'
+
+const KNOWN_MINER_ID = 'f0142637'
+
+test('get peer id of a known miner', async () => {
+  const result = await getMinerPeerId(KNOWN_MINER_ID)
+  assertMatch(result, /^12D3KooW/)
+})
+
+test('get peer id of a miner that does not exist', async () => {
+  try {
+    const result = await getMinerPeerId('f010')
+    throw new AssertionError(
+      `Expected "getMinerPeerId()" to fail, but it resolved with "${result}" instead.`
+    )
+  } catch (err) {
+    assertMatch(err.toString(), /\bf010\b.*\bactor code is not miner/)
+  }
+})