feat: fetch payload CIDs from piece indexer

backend/bin/deal-observer-backend.js

@@ -8,6 +8,7 @@ import '../lib/instrument.js'
 import { createInflux } from '../lib/telemetry.js'
 import { getChainHead, rpcRequest } from '../lib/rpc-service/service.js'
 import { fetchDealWithHighestActivatedEpoch, countStoredActiveDeals, observeBuiltinActorEvents } from '../lib/deal-observer.js'
+import { indexPieces } from '../lib/piece-indexer.js'
 import { findAndSubmitUnsubmittedDeals, submitDealsToSparkApi } from '../lib/spark-api-submit-deals.js'
 
 const {
@@ -23,8 +24,7 @@ if (!INFLUXDB_TOKEN) {
 assert(SPARK_API_BASE_URL, 'SPARK_API_BASE_URL required')
 assert(SPARK_API_TOKEN, 'SPARK_API_TOKEN required')
 
-const OBSERVE_ACTOR_EVENTS_LOOP_INTERVAL = 10 * 1000
-const SPARK_API_SUBMIT_DEALS_LOOP_INTERVAL = 10 * 1000
+const LOOP_INTERVAL = 10 * 1000
 
 // Filecoin will need some epochs to reach finality.
 // We do not want to fetch deals that are newer than the current chain head - 940 epochs.
@@ -34,6 +34,7 @@ const maxPastEpochs = 1999
 assert(finalityEpochs <= maxPastEpochs)
 
 const pgPool = await createPgPool()
+const queryLimit = 1000
 const { recordTelemetry } = createInflux(INFLUXDB_TOKEN)
 
 const observeActorEventsLoop = async (makeRpcRequest, pgPool) => {
@@ -69,12 +70,12 @@ const observeActorEventsLoop = async (makeRpcRequest, pgPool) => {
 
     if (INFLUXDB_TOKEN) {
       recordTelemetry(`loop_${slug(LOOP_NAME, '_')}`, point => {
-        point.intField('interval_ms', OBSERVE_ACTOR_EVENTS_LOOP_INTERVAL)
+        point.intField('interval_ms', LOOP_INTERVAL)
         point.intField('duration_ms', dt)
       })
     }
-    if (dt < OBSERVE_ACTOR_EVENTS_LOOP_INTERVAL) {
-      await timers.setTimeout(OBSERVE_ACTOR_EVENTS_LOOP_INTERVAL - dt)
+    if (dt < LOOP_INTERVAL) {
+      await timers.setTimeout(LOOP_INTERVAL - dt)
     }
   }
 }
@@ -107,17 +108,45 @@ const sparkApiSubmitDealsLoop = async (pgPool, { sparkApiBaseUrl, sparkApiToken,
 
     if (INFLUXDB_TOKEN) {
       recordTelemetry(`loop_${slug(LOOP_NAME, '_')}`, point => {
-        point.intField('interval_ms', SPARK_API_SUBMIT_DEALS_LOOP_INTERVAL)
+        point.intField('interval_ms', LOOP_INTERVAL)
         point.intField('duration_ms', dt)
       })
     }
-    if (dt < OBSERVE_ACTOR_EVENTS_LOOP_INTERVAL) {
-      await timers.setTimeout(SPARK_API_SUBMIT_DEALS_LOOP_INTERVAL - dt)
+    if (dt < LOOP_INTERVAL) {
+      await timers.setTimeout(LOOP_INTERVAL - dt)
+    }
+  }
+}
+
+export const pieceIndexerLoop = async (makeRpcRequest, makePixRequest, pgPool) => {
+  const LOOP_NAME = 'Piece Indexer'
+  while (true) {
+    const start = Date.now()
+    try {
+      indexPieces(makeRpcRequest, makePixRequest, pgPool, queryLimit)
+    } catch (e) {
+      console.error(e)
+      Sentry.captureException(e)
+    }
+    const dt = Date.now() - start
+    console.log(`Loop "${LOOP_NAME}" took ${dt}ms`)
+
+    // For local monitoring and debugging, we can omit sending data to InfluxDB
+    if (INFLUXDB_TOKEN) {
+      recordTelemetry(`loop_${slug(LOOP_NAME, '_')}`, point => {
+        point.intField('interval_ms', LOOP_INTERVAL)
+        point.intField('duration_ms', dt)
+      })
+    }
+    if (dt < LOOP_INTERVAL) {
+      await timers.setTimeout(LOOP_INTERVAL - dt)
     }
   }
 }
 
 await Promise.all([
+  // TODO: Define `pixRequest`
+  pieceIndexerLoop(rpcRequest, () => {}, pgPool),
   observeActorEventsLoop(rpcRequest, pgPool),
   sparkApiSubmitDealsLoop(pgPool, {
     sparkApiBaseUrl: SPARK_API_BASE_URL,

backend/lib/config.js

@@ -1,5 +1,6 @@
 const {
   RPC_URLS = 'https://api.node.glif.io/rpc/v0',
+  PIECE_INDEXER_URL = 'https://pix.filspark.com',
   GLIF_TOKEN
 } = process.env
 
@@ -15,5 +16,6 @@ if (RPC_URL.includes('glif') && GLIF_TOKEN) {
 
 export {
   RPC_URL,
-  rpcHeaders
+  rpcHeaders,
+  PIECE_INDEXER_URL
 }

backend/lib/deal-observer.js

@@ -1,11 +1,10 @@
 /** @import {Queryable} from '@filecoin-station/deal-observer-db' */
-/** @import { BlockEvent } from './rpc-service/data-types.js' */
 /** @import { Static } from '@sinclair/typebox' */
 
-import assert from 'node:assert'
 import { getActorEvents, getActorEventsFilter } from './rpc-service/service.js'
 import { ActiveDealDbEntry } from '@filecoin-station/deal-observer-db/lib/types.js'
 import { Value } from '@sinclair/typebox/value'
+import { convertBlockEventToActiveDealDbEntry } from './utils.js'
 
 /**
  * @param {number} blockHeight
@@ -15,10 +14,9 @@ import { Value } from '@sinclair/typebox/value'
  */
 export async function observeBuiltinActorEvents (blockHeight, pgPool, makeRpcRequest) {
   const eventType = 'claim'
-  const activeDeals = await getActorEvents(getActorEventsFilter(blockHeight, eventType), makeRpcRequest)
-  assert(activeDeals !== undefined, `No ${eventType} events found in block ${blockHeight}`)
-  console.log(`Observed ${activeDeals.length} ${eventType} events in block ${blockHeight}`)
-  await storeActiveDeals(activeDeals, pgPool)
+  const blockEvents = await getActorEvents(getActorEventsFilter(blockHeight, eventType), makeRpcRequest)
+  console.log(`Observed ${blockEvents.length} ${eventType} events in block ${blockHeight}`)
+  await storeActiveDeals(blockEvents.map((event) => convertBlockEventToActiveDealDbEntry(event)), pgPool)
 }
 
 /**
@@ -42,26 +40,14 @@ export async function countStoredActiveDeals (pgPool) {
 }
 
 /**
- * @param {Static<typeof BlockEvent>[]} activeDeals
+ * @param {Static<typeof ActiveDealDbEntry >[]} activeDeals
  * @param {Queryable} pgPool
  * @returns {Promise<void>}
  * */
 export async function storeActiveDeals (activeDeals, pgPool) {
-  const transformedDeals = activeDeals.map((deal) => (
-    {
-      activated_at_epoch: deal.height,
-      miner_id: deal.event.provider,
-      client_id: deal.event.client,
-      piece_cid: deal.event.pieceCid,
-      piece_size: deal.event.pieceSize,
-      term_start_epoch: deal.event.termStart,
-      term_min: deal.event.termMin,
-      term_max: deal.event.termMax,
-      sector_id: deal.event.sector,
-      payload_cid: null
-    }))
-
-  const startInserting = Date.now()
+  if (activeDeals.length === 0) {
+    return
+  }
   try {
     // Insert deals in a batch
     const insertQuery = `
@@ -89,34 +75,35 @@ export async function storeActiveDeals (activeDeals, pgPool) {
         )
       `
     await pgPool.query(insertQuery, [
-      transformedDeals.map(deal => deal.activated_at_epoch),
-      transformedDeals.map(deal => deal.miner_id),
-      transformedDeals.map(deal => deal.client_id),
-      transformedDeals.map(deal => deal.piece_cid),
-      transformedDeals.map(deal => deal.piece_size),
-      transformedDeals.map(deal => deal.term_start_epoch),
-      transformedDeals.map(deal => deal.term_min),
-      transformedDeals.map(deal => deal.term_max),
-      transformedDeals.map(deal => deal.sector_id)
+      activeDeals.map(deal => deal.activated_at_epoch),
+      activeDeals.map(deal => deal.miner_id),
+      activeDeals.map(deal => deal.client_id),
+      activeDeals.map(deal => deal.piece_cid),
+      activeDeals.map(deal => deal.piece_size),
+      activeDeals.map(deal => deal.term_start_epoch),
+      activeDeals.map(deal => deal.term_min),
+      activeDeals.map(deal => deal.term_max),
+      activeDeals.map(deal => deal.sector_id)
     ])
-
-    // Commit the transaction if all inserts are successful
-    console.log(`Inserting ${activeDeals.length} deals took ${Date.now() - startInserting}ms`)
   } catch (error) {
     // If any error occurs, roll back the transaction
-    // TODO: Add sentry entry for this error
-    // https://github.com/filecoin-station/deal-observer/issues/28
-    console.error('Error inserting deals:', error.message)
+    throw Error('Error inserting deals', { cause: error })
   }
 }
 
 /**
- * @param {Queryable} pgPool
- * @param {string} query
- * @returns {Promise<Array<Static<typeof ActiveDealDbEntry>>>}
- */
-async function loadDeals (pgPool, query) {
+   * @param {Queryable} pgPool
+   * @param {string} query
+   * @returns {Promise<Array<Static <typeof ActiveDealDbEntry>>>}
+   */
+export async function loadDeals (pgPool, query) {
   const result = (await pgPool.query(query)).rows.map(deal => {
+    // SQL used null, typebox needs undefined for null values
+    Object.keys(deal).forEach(key => {
+      if (deal[key] === null) {
+        deal[key] = undefined
+      }
+    })
     return Value.Parse(ActiveDealDbEntry, deal)
   }
   )

backend/lib/piece-indexer.js

@@ -0,0 +1,82 @@
+import { fetchPayloadCid } from './pix-service/service.js'
+import { loadDeals } from './deal-observer.js'
+import assert from 'node:assert'
+import * as util from 'node:util'
+
+/** @import {Queryable} from '@filecoin-station/deal-observer-db' */
+/** @import { Static } from '@sinclair/typebox' */
+/** @import { ActiveDealDbEntry } from '@filecoin-station/deal-observer-db/lib/types.js' */
+
+/**
+ * @param {Queryable} pgPool
+ * @param {function} makeRpcRequest
+ * @param {function} makePixRequest
+ * @returns {Promise<void>}
+ */
+export async function updatePayloadCids (pgPool, makeRpcRequest, activeDeals, makePixRequest) {
+  const updatedDeals = []
+  for (const deal of activeDeals) {
+    const payloadCid = await fetchPayloadCid(deal.miner_id, deal.piece_cid, makeRpcRequest, makePixRequest)
+    deal.payload_cid = payloadCid
+    updatedDeals.push(deal)
+  }
+  await updatePayloadInActiveDeal(pgPool, updatedDeals)
+}
+
+/**
+ *
+ * @param {function} makeRpcRequest
+ * @param {function} makePixRequest
+ * @param {Queryable} pgPool
+ * @param {number} queryLimit
+ * @returns {Promise<void>}
+ */
+export const indexPieces = async (makeRpcRequest, makePixRequest, pgPool, queryLimit) => {
+  // TODO: handle payloads which cannot be retrieved from the piece CID indexer
+  const dealsWithMissingPayloadCid = await fetchDealsWithNoPayloadCid(pgPool, queryLimit)
+  if (dealsWithMissingPayloadCid !== null && dealsWithMissingPayloadCid) {
+    await updatePayloadCids(pgPool, makeRpcRequest, dealsWithMissingPayloadCid, makePixRequest)
+  }
+}
+
+/**
+   * @param {Queryable} pgPool
+   * @param {number} limit
+   * @returns {Promise<Array<Static< typeof ActiveDealDbEntry>>>}
+   */
+export async function fetchDealsWithNoPayloadCid (pgPool, limit) {
+  assert(typeof limit === 'number', 'limit must be a number')
+  const query = `SELECT * FROM active_deals WHERE payload_cid IS NULL ORDER BY activated_at_epoch ASC LIMIT ${limit}`
+  return await loadDeals(pgPool, query)
+}
+
+/**
+ * @param {Queryable} pgPool
+ * @param {Array<Static<typeof ActiveDealDbEntry>>} deals
+ * @returns { Promise<void>}
+ */
+export async function updatePayloadInActiveDeal (pgPool, deals) {
+  const updateQuery = `
+      UPDATE active_deals
+      SET payload_cid = $1
+      WHERE activated_at_epoch = $2 AND miner_id = $3 AND client_id = $4 AND piece_cid = $5 AND piece_size = $6 AND term_start_epoch = $7 AND term_min = $8 AND term_max = $9 AND sector_id = $10
+    `
+  for (const deal of deals) {
+    try {
+      await pgPool.query(updateQuery, [
+        deal.payload_cid,
+        deal.activated_at_epoch,
+        deal.miner_id,
+        deal.client_id,
+        deal.piece_cid,
+        deal.piece_size,
+        deal.term_start_epoch,
+        deal.term_min,
+        deal.term_max,
+        deal.sector_id
+      ])
+    } catch (error) {
+      throw Error(util.format('Error updating payload of deal: ', deal), { cause: error })
+    }
+  }
+}

backend/lib/pix-service/data-types.js

@@ -0,0 +1,10 @@
+import { Type } from '@sinclair/typebox'
+
+// The response type from the piece indexer api
+const PixResponse = Type.Object({
+  samples: Type.Array(Type.String())
+})
+
+export {
+  PixResponse
+}

backend/lib/pix-service/service.js

@@ -0,0 +1,43 @@
+import { Value } from '@sinclair/typebox/value'
+import { PIECE_INDEXER_URL } from '../config.js'
+import { getMinerPeerId } from '../rpc-service/service.js'
+import { PixResponse } from './data-types.js'
+import pRetry from 'p-retry'
+
+/**
+ * @param {string} providerId
+ * @param {string} pieceCid
+ * @returns {Promise<string>}
+ */
+export const getDealPayloadCid = async (providerId, pieceCid) => {
+  const url = PIECE_INDEXER_URL + '/sample/' + providerId + '/' + pieceCid
+  try {
+    const response = await pRetry(async () => await fetch(url, {
+      method: 'GET',
+      headers: { 'content-type': 'application/json' }
+    }), { retries: 5 })
+    const json = await response.json()
+    try {
+      const parsedPixResponse = Value.Parse(PixResponse, json)
+      if (parsedPixResponse.samples.length === 0) {
+        throw new Error('No samples found in the response.')
+      }
+      return parsedPixResponse.samples[0]
+    } catch (e) {
+      throw new Error(`Failed to parse response from piece indexer. The response was : ${JSON.stringify(json)}`, { cause: e })
+    }
+  } catch (e) {
+    throw new Error('Failed to make RPC request.', { cause: e })
+  }
+}
+
+/**
+* @param {number} providerId
+* @param {string} pieceCid
+* @returns {Promise<string>}
+*/
+export async function fetchPayloadCid (providerId, pieceCid, makeRpcRequest, makePixRequest) {
+  const minerPeerId = await getMinerPeerId(providerId, makeRpcRequest)
+  const payloadCid = await makePixRequest(minerPeerId, pieceCid)
+  return payloadCid
+}